Gleitende Fenster - Entwicklerhandbuch für Amazon Kinesis Data Analytics for SQL Applications
Beispiel 1Beispiel 2: Abfrage, die Zusammenfassungen auf ein gleitendes Fenster anwendetBeispiel 3: Abfragen von Daten aus mehreren gleitenden Fenstern für denselben Stream

Für neue Projekte empfehlen wir, den neuen Managed Service für Apache Flink Studio anstelle von Kinesis Data Analytics for SQL Applications zu verwenden. Der Managed Service für Apache Flink Studio kombiniert Benutzerfreundlichkeit mit fortschrittlichen Analysefunktionen, sodass Sie in wenigen Minuten anspruchsvolle Anwendungen zur Stream-Verarbeitung erstellen können.

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Gleitende Fenster

Statt Datensätze mittels GROUP BY zu gruppieren, können Sie ein Fenster auf Zeit- oder Zeilenbasis definieren. Sie führen dies aus, indem Sie eine explizite WINDOW-Klausel hinzufügen.

In diesem Fall gibt Amazon Kinesis Data Analytics eine Ausgabe aus, während das Fenster über die Zeit gleitet und neue Datensätze im Stream angezeigt werden. Kinesis Data Analytics gibt diese Ausgabe aus, indem Zeilen im Fenster verarbeitet werden. Bei der Art der Verarbeitung können sich Fenster überschneiden; ein Datensatz kann Teil mehrerer Fenster sein und wird mit jeden Fenster verarbeitet. Das folgende Beispiel zeigt ein gleitendes Fenster.

Betrachten Sie eine einfache Abfrage, die Datensätze im Stream zählt. Dieses Beispiel setzt ein 5-Sekunden-Fenster voraus. Im folgenden Beispiel-Stream treffen neue Datensätze zu den Zeitpunkten t1, t2, t6 und t7 ein. Drei Datensätze treffen zum Zeitpunkt t8 Sekunden ein.

Timeline showing record arrivals at t1, t2, t6, t7, and multiple at t8 within a 5-second window.

Beachten Sie Folgendes:

  • Das Beispiel setzt ein 5-Sekunden-Fenster voraus. Das Fenster von 5 Sekunden gleitet kontinuierlich über die Zeit.

  • Für jede Zeile, die in ein Fenster eintritt, wird vom gleitenden Fenster eine Ausgabezeile ausgegeben. Direkt nach dem Start der Anwendung wird Ihnen die Abfrageausgabe für jeden neuen Datensatz angezeigt, der im Stream angezeigt wird, auch wenn ein bestimmtes 5-Sekunden-Fenster noch nicht verstrichen ist. Die Abfrage gibt beispielsweise eine Ausgabe aus, wenn ein Datensatz in der ersten und zweiten Sekunde eintrifft. Später verarbeitet die Abfrage Datensätze im 5-Sekunden-Fenster.

  • Die Fenster gleiten mit der Zeit. Wenn ein alter Datensatz im Stream aus dem Fenster herausfällt, gibt die Abfrage keine Ausgabe aus, es sei denn, es gibt auch einen neuen Datensatz im Stream, der in dieses 5-Sekunden-Fenster fällt.

Angenommen, die Abfrage wird zum Zeitpunkt t0 gestartet. Dann geschieht Folgendes:

  1. Zum Zeitpunkt t0 startet die Abfrage. Die Abfrage gibt keine Ausgabe (Zählwert) aus, da es zu diesem Zeitpunkt keine Datensätze gibt.

    Timeline showing a stream starting at t0 with no output initially indicated.

  2. Zum Zeitpunkt t1 trifft ein neuer Datensatz im Stream ein und die Abfrage gibt den Zählwert 1 aus.

    Timeline showing a stream with a record appearing at time t1, and an arrow pointing to t0.

  3. Zum Zeitpunkt t2 trifft ein weiterer Datensatz ein und die Abfrage gibt den Zählwert 2 aus.

    Timeline showing stream events at different time points, with two vertical bars at the end.

  4. Das 5-Sekunden-Fenster gleitet mit der Zeit:

    • Zum Zeitpunkt t3, das gleitende Fenster von t3 zu t0

    • Zum Zeitpunkt t4 (das gleitende Fenster von t4 zu t0)

    • Zum Zeitpunkt t5, das gleitetende Fenster von t5 zu t0.

    Zu all diesen Zeitpunkten besitzt das 5-Sekunden-Fenster dieselben Datensätze – es gibt keine neuen Datensätze. Daher gibt die Abfrage keine Ausgabe aus.

    Timeline showing stream with multiple time points and colored rectangles representing data windows.

  5. Zum Zeitpunkt t6 liegt das 5-Sekunden-Fenster zwischen t6 und t1. Die Abfrage erkennt einen einzelnen neuen Datensatz bei t6 und gibt daher als Ausgabe 2 aus. Der Datensatz bei t1 liegt nicht mehr im Fenster und wird nicht gezählt.

    Timeline showing stream events at different time points with a sliding 5-second window.

  6. Zum Zeitpunkt t7 liegt das 5-Sekunden-Fenster zwischen t7 und t2. Die Abfrage erkennt einen einzelnen neuen Datensatz bei t7 und gibt daher als Ausgabe 2 aus. Der Datensatz liegt zum Zeitpunkt t2 nicht mehr im 5-Sekunden-Fenster und wird daher nicht gezählt.

    Timeline showing stream events and time points from t0 to t7, with a 5-second window highlighted.

  7. Zum Zeitpunkt t8 liegt das 5-Sekunden-Fenster zwischen t8 und t3. Die Abfrage erkennt drei neue Datensätze und gibt daher die Datensatzzahl 5 aus.

    Timeline showing stream events with orange bars representing record counts at different time intervals.

Zusammenfassend kann gesagt werden, dass das Fenster eine feste Größe hat und mit der Zeit gleitet. Die Abfrage gibt eine Ausgabe aus, wenn neue Datensätze angezeigt werden.

Anmerkung

Wir empfehlen, keine gleitenden Fenster zu verwenden, die länger als eine Stunde sind. Wenn Sie ein längeres Fenster verwenden, nimmt der Neustart der Anwendung nach einer regulären Systemwartung mehr Zeit in Anspruch. Dies liegt daran, dass die Quelldaten erneut aus dem Stream gelesen werden müssen.

In den folgenden Beispielabfragen wird die WINDOW-Klausel verwendet, um Fenster zu definieren und Zusammenfassungen auszuführen. Da die Abfragen GROUP BY nicht angeben, verwendet die Abfrage für die Verarbeitung von Datensätzen im Stream gleitende Fenster.

Beispiel 1: Verarbeiten eines Streams mit einem gleitenden 1-Minuten-Fenster

Betrachten Sie den Demo-Stream in der Einführungsübung, der den In-Application-Stream SOURCE_SQL_STREAM_001 auffüllt. Im Folgenden wird das Schema gezeigt.

(TICKER_SYMBOL VARCHAR(4), 
 SECTOR varchar(16),
 CHANGE REAL,
 PRICE REAL)

Angenommen, Ihre Anwendung soll Zusammenfassungen mithilfe eines gleitenden 1-Minuten-Fensters verarbeiten. Das bedeutet, dass Ihre Anwendung für jeden neuen, im Stream angezeigten Datensatz eine Ausgabe ausgeben soll, indem Datensätze im vorhergehenden 1-Minuten-Fenster zusammengefasst werden.

Sie können die folgende Abfrage mit Fenster auf Zeitbasis verwenden. Die Abfrage verwendet die WINDOW-Klausel, um das 1-Minuten-Bereichsintervall zu definieren. Die Angabe PARTITION BY in der WINDOW-Klausel gruppiert Datensätze nach Tickerwerten innerhalb des gleitenden Fensters. 

SELECT STREAM ticker_symbol,
              MIN(Price) OVER W1 AS Min_Price,
              MAX(Price) OVER W1 AS Max_Price,
              AVG(Price) OVER W1 AS Avg_Price
FROM   "SOURCE_SQL_STREAM_001"
WINDOW W1 AS (
   PARTITION BY ticker_symbol 
   RANGE INTERVAL '1' MINUTE PRECEDING);
So testen Sie die Abfrage

  1. Richten Sie anhand der Erste Schritte-Übung eine Anwendung ein.

  2. Ersetzen Sie die SELECT-Anweisung im Anwendungscode durch die vorherige SELECT-Abfrage. Der resultierende Anwendungscode wird im Folgenden gezeigt.

    CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
                         ticker_symbol VARCHAR(10), 
                         Min_Price     double, 
                         Max_Price     double, 
                         Avg_Price     double);
CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
   INSERT INTO "DESTINATION_SQL_STREAM"
     SELECT STREAM ticker_symbol,
                   MIN(Price) OVER W1 AS Min_Price,
                   MAX(Price) OVER W1 AS Max_Price,
                   AVG(Price) OVER W1 AS Avg_Price
     FROM   "SOURCE_SQL_STREAM_001"
     WINDOW W1 AS (
        PARTITION BY ticker_symbol 
        RANGE INTERVAL '1' MINUTE PRECEDING);

Beispiel 2: Abfrage, die Zusammenfassungen auf ein gleitendes Fenster anwendet

Die folgende Abfrage für den Demo-Stream gibt die durchschnittliche prozentuale Änderung in Bezug auf den Preis der einzelnen Ticker in einem 10-Sekunden-Fenster zurück.

SELECT STREAM Ticker_Symbol,
              AVG(Change / (Price - Change)) over W1 as Avg_Percent_Change
FROM "SOURCE_SQL_STREAM_001"
WINDOW W1 AS (
        PARTITION BY ticker_symbol 
        RANGE INTERVAL '10' SECOND PRECEDING);

So testen Sie die Abfrage

  1. Richten Sie anhand der Erste Schritte-Übung eine Anwendung ein.

  2. Ersetzen Sie die SELECT-Anweisung im Anwendungscode durch die vorherige SELECT-Abfrage. Der resultierende Anwendungscode wird im Folgenden gezeigt.

    CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
                            ticker_symbol VARCHAR(10), 
                            Avg_Percent_Change double);
CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
   INSERT INTO "DESTINATION_SQL_STREAM"
      SELECT STREAM Ticker_Symbol,
                    AVG(Change / (Price - Change)) over W1 as Avg_Percent_Change
      FROM "SOURCE_SQL_STREAM_001"
      WINDOW W1 AS (
              PARTITION BY ticker_symbol 
              RANGE INTERVAL '10' SECOND PRECEDING);

Beispiel 3: Abfragen von Daten aus mehreren gleitenden Fenstern für denselben Stream

Sie können Abfragen schreiben, um Ausgaben auszugeben, in denen jeder Spaltenwert mittels unterschiedlicher gleitender Fenster berechnet wird, die für denselben Stream definiert sind.

Im folgenden Beispiel gibt die Abfrage den Ausgabe Ticker, den Preis, a2 und a10 aus. Sie gibt eine Ausgabe für Ticker-Symbole aus, deren gleitender Zwei-Zeilen-Durchschnitt den gleitenden Zehn-Zeilen-Durchschnitt überschreitet. Die Spaltenwerte a2 und a10 werden aus Fenstern abgeleitet, die über zwei Zeilen und zehn Zeilen gleiten.

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
                           ticker_symbol      VARCHAR(12), 
                           price              double, 
                           average_last2rows  double, 
                           average_last10rows double);

CREATE OR REPLACE PUMP "myPump" AS INSERT INTO "DESTINATION_SQL_STREAM"
SELECT STREAM ticker_symbol, 
              price, 
              avg(price) over last2rows, 
              avg(price) over last10rows
FROM SOURCE_SQL_STREAM_001
WINDOW
    last2rows AS (PARTITION BY ticker_symbol ROWS 2 PRECEDING),
    last10rows AS (PARTITION BY ticker_symbol ROWS 10 PRECEDING);

Um diese Abfrage anhand des Demo-Streams zu testen, befolgen Sie das in Beispiel 1 beschriebene Testverfahren.