在仔細考慮之後，我們決定停止 Amazon Kinesis Data Analytics for SQL 應用程式： 1. 從 **2025 年 9 月 1 日起，**我們不會為 Amazon Kinesis Data Analytics for SQL 應用程式提供任何錯誤修正，因為考慮到即將終止，我們將對其提供有限的支援。 2. 從 **2025 年 10 月 15 日起，**您將無法建立新的 Kinesis Data Analytics for SQL 應用程式。 3. 我們將自 **2026 年 1 月 27** 日起刪除您的應用程式。您將無法啟動或操作 Amazon Kinesis Data Analytics for SQL 應用程式。從那時起，Amazon Kinesis Data Analytics for SQL 將不再提供支援。如需詳細資訊，請參閱[Amazon Kinesis Data Analytics for SQL 應用程式終止](discontinuation.md)。 本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 串流 SQL 概念 Amazon Kinesis Data Analytics 會使用延伸模組實作 ANSI 2008 SQL 標準。這些延伸模組可讓您處理串流資料。下列主題涵蓋重要的串流 SQL 概念。 **Topics** + [應用程式內串流與幫浦](streams-pumps.md) + [時間戳記和 ROWTIME 欄](timestamps-rowtime-concepts.md) + [持續查詢](continuous-queries-concepts.md) + [窗口化查詢](windowed-sql.md) + [串流資料作業：串流聯結](stream-joins-concepts.md) # 應用程式內串流與幫浦 在設定[應用程式輸入](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/how-it-works-input.html)時，您會將串流來源映射到建立的應用程式內串流。資料會持續從串流來源流入應用程式內串流。應用程式內串流的運作方式，與您可以使用 SQL 陳述式查詢的資料表類似，但這稱為串流，因為其代表連續的資料流程。 **注意** 請勿將應用程式內串流與 Amazon Kinesis 資料串流和 Firehose 交付串流混淆。應用程式內串流僅存在 Amazon Kinesis Data Analytics 應用程式的環境中。Kinesis 資料串流和 Firehose 交付串流與您的應用程式無關。您可以在應用程式輸入組態將它們設定為串流來源，或在輸出組態將其設定為目的地。 您也可以視需要建立更多應用程式內串流，以儲存中繼查詢結果。建立應用程式內串流分為兩個步驟。首先，建立應用程式內串流，然後將資料送入其中。例如，假設應用程式的輸入組態會建立名 `INPUTSTREAM` 為的應用程式內串流。在下列範例中，您會建立另一個串流 (`TEMPSTREAM`)，然後送進從 `INPUTSTREAM` 抽取的資料。 1. 建立具有三個資料欄的應用程式內串流 (`TEMPSTREAM`)，如下所示： ``` CREATE OR REPLACE STREAM "TEMPSTREAM" ( "column1" BIGINT NOT NULL, "column2" INTEGER, "column3" VARCHAR(64)); ``` 資料欄名稱位於引號中，代表其區分大小寫。如需詳細資訊，請參閱《Amazon Kinesis Data Analytics SQL 參考》**中的[識別符](https://docs.aws.amazon.com/kinesisanalytics/latest/sqlref/sql-reference-identifiers.html)。 1. 使用幫浦將數據插入流中。幫浦是執行中的連續插入查詢，可將資料從一個應用程式內串流插入另一個應用程式內串流。下列陳述式會建立幫浦 (`SAMPLEPUMP`)，並透過從另一個資料流 (`INPUTSTREAM`) 選取記錄，將資料插入 `TEMPSTREAM` 中。 ``` CREATE OR REPLACE PUMP "SAMPLEPUMP" AS INSERT INTO "TEMPSTREAM" ("column1", "column2", "column3") SELECT STREAM inputcolumn1, inputcolumn2, inputcolumn3 FROM "INPUTSTREAM"; ``` 您可以讓多個寫入器插入應用程式內串流，而且可以從串流中選取多個讀取器。將應用程式內串流視為實作發佈／訂閱訊息範例。在此範例中，資料列 (包括建立時間和接收時間) 可以透過串流 SQL 陳述式的串聯處理、解譯和轉送，而不必儲存在傳統的 RDBMS 中。 建立應用程式內串流之後，您可以執行一般 SQL 查詢。 **注意** 當您查詢串流時，大多數 SQL 陳述式都會使用資料列或時間型窗口繫結。如需詳細資訊，請參閱[窗口化查詢](windowed-sql.md)。 您也可以加入串流。如需加入串流的範例，請參閱 [串流資料作業：串流聯結](stream-joins-concepts.md)。 # 時間戳記和 ROWTIME 欄 應用程式內串流包含一個名為 `ROWTIME` 的特殊欄。當 Amazon Kinesis Data Analytics 在第一個應用程式內串流中插入資料列時，它會儲存時間戳記。`ROWTIME` 反映 Amazon Kinesis Data Analytics 從串流來源讀取後，將記錄插入第一個應用程式內串流的時間戳記。然後此 `ROWTIME` 值會在整個應用程式中保留。 **注意** 當您將記錄從一個應用程式內串流抽取到另一個應用程式內串流時，不需要明確複製該 `ROWTIME` 欄，Amazon Kinesis Data Analytics 會為您複製此資料欄。 Amazon Kinesis Data Analytics 保證這些 `ROWTIME` 值會單調增加。您可以在基於時間的窗口式查詢中使用此時間戳記。如需詳細資訊，請參閱[窗口化查詢](windowed-sql.md)。 您可以像在應用程式內串流中存取任何其他資料欄一樣，存取 `SELECT` 陳述式中的 ROWTIME 資料欄。例如： ``` SELECT STREAM ROWTIME, some_col_1, some_col_2 FROM SOURCE_SQL_STREAM_001 ``` ## 了解串流分析中的不同時間 除了 `ROWTIME` 之外，實時串流應用程式中還有其他類型的時間。這些時間為： + **事件時間**：事件發生的時間戳記。這有時也稱為*用戶端時間*。在分析中通常偏好使用此時間，因其為事件發生的時間。不過，許多事件來源 (例如行動電話和 Web 用戶端) 沒有可靠的時鐘，這可能會導致不正確的時間。此外，連線問題可能會導致串流上的記錄顯示順序與事件發生的順序不相同。   + **擷取時間**：記錄新增至串流來源的時間戳記。Amazon Kinesis Data Streams 在提供此時間戳記的每個記錄中，都包含一個名為 `APPROXIMATE_ARRIVAL_TIME` 的欄位。這有時也稱為*伺服器端時間*。這個擷取時間通常是接近事件時間的近似值。如果記錄擷取到串流時有任何延遲，則可能會導致錯誤，但通常很少見。此外，擷取時間很少出現故障，但由於串流資料的分散式性質，故障可能會發生。因此，擷取時間大多準確且符合順序地反映事件時間。   + **處理時間**：Amazon Kinesis Data Analytics 在第一個應用程式內串流中插入資料列的時間戳記。在每個應用程式內串流中，Amazon Kinesis Data Analytics 會在 `ROWTIME` 欄提供時間戳記。處理時間總是會單調增加。但是，如果您的應用程式落後，此時間就不准確。(如果應用程式落後，處理時間就不能準確反映事件時間。) 此 `ROWTIME` 與掛鐘比對是準確的，但它可能不是事件實際發生的時間。 在基於時間的窗口查詢中，使用這些時間中的每一個時間都有優點和缺點。我們建議您選擇其中一個或多個時間，並擬好根據使用案例情境處理相關缺點的策略。 **注意** 如果您使用以列為基礎的窗口，則時間不是問題，您可以忽略此節。 我們建議採用雙窗口策略，該策略使用兩個時間基礎，即 `ROWTIME` 與另一個其他時間 (擷取或事件時間) 兩者。 + 使用 `ROWTIME` 作為第一個窗口，此時段可控制查詢發出結果的頻率，如下列範例所示。這並非邏輯時間。 + 把其中一個其他時間當作邏輯時間，即您想要連結到分析的時間 此時間表示事件發生的時間。在下面的例子中，分析目標是按股票代號對記錄進行分組和返回計數。 此策略的優點，是可以使用代表事件發生的時間。當您的應用程式落後或事件出現故障時，它可以適當地處理。如果應用程式在將記錄引入應用程式內串流時落後，它們仍會依照第二個窗口中的邏輯時間進行分組。該查詢用 `ROWTIME` 來保證處理的順序。任何延遲的記錄 (與 `ROWTIME` 值相比，擷取時間戳記顯示較早的值) 也會成功處理。 請考慮下列針對[入門練習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)中使用的示範串流的查詢。該查詢使用 `GROUP BY` 子句，並在一分鐘的輪轉窗口中發出股票代號計數。 ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ("ingest_time" timestamp, "APPROXIMATE_ARRIVAL_TIME" timestamp, "ticker_symbol" VARCHAR(12), "symbol_count" integer); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND) AS "ingest_time", STEP("SOURCE_SQL_STREAM_001".APPROXIMATE_ARRIVAL_TIME BY INTERVAL '60' SECOND) AS "APPROXIMATE_ARRIVAL_TIME", "TICKER_SYMBOL", COUNT(*) AS "symbol_count" FROM "SOURCE_SQL_STREAM_001" GROUP BY "TICKER_SYMBOL", STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND), STEP("SOURCE_SQL_STREAM_001".APPROXIMATE_ARRIVAL_TIME BY INTERVAL '60' SECOND); ``` 在 `GROUP BY`，您先依據 `ROWTIME` 將記錄分組在一分鐘的窗口中，然後再依據 `APPROXIMATE_ARRIVAL_TIME`。 結果中的時間戳記值會無條件捨去至最接近的 60 秒間隔。查詢發出的第一個群組結果，會顯示第一分鐘的記錄。發出的第二組結果，會根據 `ROWTIME` 顯示接下來幾分鐘內的記錄。最後一筆記錄指出應用程式擷取紀錄至應用程式內串流的時間延遲 (與擷取時間戳記相比，顯示了延遲 `ROWTIME` 值)。 ``` ROWTIME INGEST_TIME TICKER_SYMBOL SYMBOL_COUNT --First one minute window. 2016-07-19 17:05:00.0 2016-07-19 17:05:00.0 ABC 10 2016-07-19 17:05:00.0 2016-07-19 17:05:00.0 DEF 15 2016-07-19 17:05:00.0 2016-07-19 17:05:00.0 XYZ 6 –-Second one minute window. 2016-07-19 17:06:00.0 2016-07-19 17:06:00.0 ABC 11 2016-07-19 17:06:00.0 2016-07-19 17:06:00.0 DEF 11 2016-07-19 17:06:00.0 2016-07-19 17:05:00.0 XYZ 1 *** ***late-arriving record, instead of appearing in the result of the first 1-minute windows (based on ingest_time, it is in the result of the second 1-minute window. ``` 透過將結果推送至下游資料庫，您可以合併結果，以獲得最終精確的每分鐘計數。例如，您可以設定應用程式輸出，將結果保留到可寫入 Amazon Redshift 資料表的 Firehose 交付串流。在 Amazon Redshift 資料表中顯示結果之後，您可以查詢資料表來計算依據 `Ticker_Symbol` 分組的總計數。在 `XYZ` 的情況，即使記錄遲到，總數也是準確的（6\$11）。 # 持續查詢 串流上的查詢會透過串流資料持續執行。這種持續執行可啟用案例，例如應用程式能夠持續查詢串流並產生提醒。 在入門練習中，您有一個名為 `SOURCE_SQL_STREAM_001` 的應用程式內串流。它會持續從示範串流 (Kinesis 資料串流) 接收股票價格。結構描述如下： ``` (TICKER_SYMBOL VARCHAR(4), SECTOR varchar(16), CHANGE REAL, PRICE REAL) ``` 假設您對超過 15％ 的股票價格變動感興趣。您可以在應用程式碼中使用下列查詢。此查詢會持續執行，並在偵測到股票價格變動大於 15% 時發出記錄。 ``` SELECT STREAM TICKER_SYMBOL, PRICE FROM "SOURCE_SQL_STREAM_001" WHERE (ABS((CHANGE / (PRICE-CHANGE)) * 100)) > 15 ``` 使用下列程序來設定 Amazon Kinesis Data Analytics 應用程式，並測試此查詢。 **若要測試查詢** 1. 按照[入門練習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)建立應用程式。 1. 用上述 `SELECT` 查詢取代應用程式碼中的 `SELECT` 陳述式。產生的應用程式碼如下所示： ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (ticker_symbol VARCHAR(4), price DOUBLE); -- CREATE OR REPLACE PUMP to insert into output CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM TICKER_SYMBOL, PRICE FROM "SOURCE_SQL_STREAM_001" WHERE (ABS((CHANGE / (PRICE-CHANGE)) * 100)) > 15; ``` # 窗口化查詢 應用程式碼中的 SQL 查詢會透過應用程式內串流持續執行。應用程式內串流，代表的是持續在應用程式中流動的無限制資料。因此，若要取得此持續更新輸入的結果集，經常會使用定義的窗口時段或資料列來限制查詢。這些也被稱為*窗口式 SQL*。 對於以時間為基礎的窗口化查詢，您可以根據時間來指定窗口大小 (例如，一分鐘的窗口)。這需要應用程式內串流中的時間戳記資料欄，該欄會單調增加。(新資料列的時間戳記大於或等於上一列。) 在每個應用程式內串流中，Amazon Kinesis Data Analytics 會提供這樣的時間戳記欄，名為 `ROWTIME`。您可以在指定基於時間的查詢中使用此資料欄。針對應用程式，您可以選擇其他時間戳記選項。如需詳細資訊，請參閱[時間戳記和 ROWTIME 欄](timestamps-rowtime-concepts.md)。 針對以資料列為基礎的窗口化查詢，您可以根據資料列數來指定窗口大小。 您可以根據應用程式需求，指定查詢以輪轉窗口、滑動窗口或交錯窗口方式處理記錄。Kinesis Data Analytics 支援下列窗口類型： + [交錯窗口](stagger-window-concepts.md)：此查詢使用金鑰式時間窗口彙總資料，該窗口會在資料到達時打開。這些金鑰允許多個重疊的窗口。這是使用以時間窗口彙總資料的建議方式，因為與輪轉窗口相比，交錯窗口會減少遲到或順序不符的資料。 + [輪轉窗口](tumbling-window-concepts.md)：此查詢使用定期開啟和關閉的時間窗口來彙總資料。 + [滑動視窗](sliding-window-concepts.md)：此查詢使用固定時間或資料列計數間隔持續彙總資料。 # 交錯窗口 使用*交錯窗口*是一種窗口化方法，適用於分析不一致時間到達的資料群組。此方式非常適合任何時間序列分析使用案例，例如一組相關的銷售或日誌記錄。 例如，[VPC 流程日誌](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html#flow-logs-limitations)的擷取窗口約為 10 分鐘。但如果你在客戶端上彙總資料，則可以有一個長達 15 分鐘的擷取窗口，。交錯視窗是彙總這些日誌以進行分析的理想選擇。 交錯窗口可解決相關記錄未落入相同時間限制窗口的問題，例如使用輪轉窗口時。 ## 有輪轉窗口的部分結果 彙總遲到或順序錯誤的資料時，使用 [輪轉窗口](tumbling-window-concepts.md) 有某些限制。 如果使用輪轉窗口來分析與時間相關的資料群組，則個別記錄可能會落入不同的窗口中。因此，每個窗口的部分結果必須在稍後合併，以產生每組記錄的完整結果。 在下面的輪轉窗口查詢中，記錄按列時間，事件時間和股票代號分組到窗口中： ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( TICKER_SYMBOL VARCHAR(4), EVENT_TIME timestamp, TICKER_COUNT DOUBLE); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM TICKER_SYMBOL, FLOOR(EVENT_TIME TO MINUTE), COUNT(TICKER_SYMBOL) AS TICKER_COUNT FROM "SOURCE_SQL_STREAM_001" GROUP BY ticker_symbol, FLOOR(EVENT_TIME TO MINUTE), STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '1' MINUTE); ``` 在下圖中，應用程式根據交易發生的時間（事件時間），以一分鐘的精細程度計算接收的交易數量。該應用程式可以使用輪轉窗口，根據列時間和事件時間幫資料分組。該應用程式收到四條記錄，這些記錄彼此都在一分鐘內到達。它按列時間，事件時間和股票符號幫記錄分組。因為有些記錄會在第一個輪轉窗口結束後到達，所以記錄不會全部落在相同的一分鐘輪轉窗口內。 ![\[Tumbling windows diagram showing data grouping by row time, event time, and ticker symbol over two minutes.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/stagger_0.png) 前面的圖表具有以下事件。 **** | ROWTIME | EVENT\$1TIME | TICKER\$1SYMBOL | | --- | --- | --- | | 11：00：20 | 11：00：10 | AMZN | | 11：00：30 | 11：00：20 | AMZN | | 11：01：05 | 11：00：55 | AMZN | | 11：01：15 | 11：01：05 | AMZN | 來自輪轉窗口應用程式的結果集類似如下。 **** | ROWTIME | EVENT\$1TIME | TICKER\$1SYMBOL | COUNT | | --- | --- | --- | --- | | 11：01：00 | 11：00：00 | AMZN | 2 | | 11：02：00 | 11：00：00 | AMZN | 1 | | 11：02：00 | 11：01：00 | AMZN | 1 | 在之前的結果集中，傳回了三個結果： + 記錄的 `ROWTIME` 為 11:01:00，彙總了前兩個記錄。 + 11：02：00 的記錄僅彙總了第三條記錄。此記錄在第二個窗口中有一個 `ROWTIME`，但在第一個窗口中有一個 `EVENT_TIME`。 + 11：02：00 的記錄僅彙總了第四條記錄。 若要分析完整的結果集，必須在持續性存放區彙總記錄。這會增加應用程式的複雜性和處理需求。 ## 交錯窗口的完整結果 為了提高分析時間相關資料記錄的準確性，Kinesis Data Analytics 提供了一種稱為*交錯窗口*的新窗口類型。在此窗口類型中，窗口會在第一個與分割區索引鍵相符的事件到達時打開，而不是在固定的時間間隔。窗口會依指定的存留期關閉，測量由窗口開啟時起算。 在窗口子句中，交錯窗口是每個索引鍵群組的個別時間限制窗口。應用程式會在自己的時間窗口內彙總窗口子句的每個結果，而不是針對所有結果使用單一窗口。 在下面的交錯窗口查詢中，記錄按事件時間和股票代號分組到窗口中： ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(4), event_time TIMESTAMP, ticker_count DOUBLE); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM TICKER_SYMBOL, FLOOR(EVENT_TIME TO MINUTE), COUNT(TICKER_SYMBOL) AS ticker_count FROM "SOURCE_SQL_STREAM_001" WINDOWED BY STAGGER ( PARTITION BY FLOOR(EVENT_TIME TO MINUTE), TICKER_SYMBOL RANGE INTERVAL '1' MINUTE); ``` 在下圖中，事件按事件時間和股票代號彙總到交錯窗口中。 ![\[Diagram showing event aggregation into stagger windows by event time and ticker symbol.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/stagger_1.png) 上圖有下列事件，這些事件輪轉窗口應用程式分析的相同： **** | ROWTIME | EVENT\$1TIME | TICKER\$1SYMBOL | | --- | --- | --- | | 11：00：20 | 11：00：10 | AMZN | | 11：00：30 | 11：00：20 | AMZN | | 11：01：05 | 11：00：55 | AMZN | | 11：01：15 | 11：01：05 | AMZN | 來自交錯窗口應用程式的結果集類似如下。 **** | ROWTIME | EVENT\$1TIME | TICKER\$1SYMBOL | 計數 | | --- | --- | --- | --- | | 11：01：20 | 11：00：00 | AMZN | 3 | | 11：02：15 | 11：01：00 | AMZN | 1 | 傳回的記錄聚合前三個輸入記錄。記錄會依一分鐘的交錯窗口來分組。當應用程式收到第一筆 AMZN 記錄 (其中 `ROWTIME` 為 11:00:20) 時，就會啟動交錯窗口。當 1 分鐘交錯窗口到期時 (11:01:20)，會將包含落在交錯窗口內 (以 `ROWTIME` 和 `EVENT_TIME` 為基礎) 的結果的記錄寫入輸出串流。如使用交錯窗口，所有在一分鐘視窗內有 `ROWTIME` 和 `EVENT_TIME` 的記錄都會在單一結果中發出。 最後一筆記錄 (在一分鐘彙總外有 `EVENT_TIME`) 會分別彙總。這是因為 `EVENT_TIME` 是用來將記錄分隔成結果集的分割區索引鍵之一，而第一個視窗的 `EVENT_TIME` 分割區索引鍵是 `11:00`。 交錯窗口的語法是在特殊子句 `WINDOWED BY` 中定義的。使用此子句，而不是用於串流聚合的 `GROUP BY` 子句。子句會緊接顯示在選用 `WHERE` 子句之後，在 `HAVING` 子句之前。 交錯窗口在 `WINDOWED BY` 子句中定義，並採用兩個參數：分割區索引鍵和窗口長度。分割區索引鍵會分割傳入的資料串流，並定義窗口開啟的時間。當串流上出現具有唯一分割區索引鍵的第一個事件時，會開啟交錯窗口。交錯窗口會在窗口長度所定義的固定期間之後關閉。語法如下列程式碼範例所示： ``` ... FROM WHERE <... optional statements...> WINDOWED BY STAGGER( PARTITION BY RANGE INTERVAL ); ``` # 輪轉窗口（使用 GROUP BY 彙總） 當窗口查詢以非重疊的方式處理每個窗口時，即稱作*輪轉窗口*。在此情況下，應用程式內串流上的每個記錄都屬於一個特定窗口。紀錄只會被處理一次（當查詢處理記錄所屬的窗口時）。 ![\[Timeline showing non-overlapping windows processing data streams at distinct time intervals.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/window-tumbling-20.png) 例如，使用 `GROUP BY` 子句的彙總查詢會處理輪轉窗口中的資料列。[入門練習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)的示範串流，會接收您的應用程式內串流 `SOURCE_SQL_STREAM_001` 之股票價格資料。這個串流具有以下結構描述： ``` (TICKER_SYMBOL VARCHAR(4), SECTOR varchar(16), CHANGE REAL, PRICE REAL) ``` 在應用程式碼中，假設您想要在一分鐘的窗口中尋找每個股票代號的彙總 (最低、最大值) 價格。您可以使用下列查詢： ``` SELECT STREAM ROWTIME, Ticker_Symbol, MIN(Price) AS Price, MAX(Price) AS Price FROM "SOURCE_SQL_STREAM_001" GROUP BY Ticker_Symbol, STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND); ``` 前面是以時間為基礎的窗口化查詢範例。此查詢按 `ROWTIME` 值為記錄分組。針對每分鐘進行報告，`STEP` 函數會將 `ROWTIME` 值無條件捨去至最接近的分鐘。 **注意** 您也可以使用 `FLOOR` 函數來將記錄分組至窗口。但是，`FLOOR` 只能將時間值捨去到完整時間單位 (小時、分鐘、秒等)。建議使用 `STEP` 將記錄分組到輪轉窗口中，因為它可以將值捨去到任意間隔，例如 30 秒。 此查詢是非重疊（輪轉）窗口的例子。`GROUP BY` 子句會將記錄分組在一分鐘的窗口中，而且每個記錄都屬於特定的窗口 (不重疊)。該查詢每分鐘發出一個輸出記錄，提供在特定分鐘記錄的最小/最大股票價格。如要從輸入資料串流生成定期報告，這種類型的查詢就非常有用。在此範例中，每分鐘產生一次報告。 **若要測試查詢** 1. 按照[入門練習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)設置應用程式。 1. 用上述 `SELECT` 查詢取代應用程式碼中的 `SELECT` 陳述式。產生的應用程式碼如下所示： ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(4), Min_Price DOUBLE, Max_Price DOUBLE); -- CREATE OR REPLACE PUMP to insert into output CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM Ticker_Symbol, MIN(Price) AS Min_Price, MAX(Price) AS Max_Price FROM "SOURCE_SQL_STREAM_001" GROUP BY Ticker_Symbol, STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND); ``` # 滑動視窗 您可以定義以時間或資料列為基礎的視窗，而不需使用 `GROUP BY` 將記錄分組。您可以透過添加明確的 `WINDOW` 子句來完成此操作。 在這種情況下，當窗口隨著時間的推移而滑動時，Amazon Kinesis Data Analytics 會在串流上出現新記錄時發出輸出。Kinesis Data Analytics 會透過處理窗口中的資料列來發出此輸出。窗口可以在這種類型的處理中重疊，且記錄可以是多個窗口的一部分，並用每個窗口進行處理。以下範例解釋了滑動窗口。 考慮用一個簡單的查詢來計算串流上的記錄。此範例假設了 5 秒的窗口。在下面的示範串流中，新的記錄到達時間為 t1、t2、t6 和 t7，三個記錄到達時間為 t8 秒。 ![\[Timeline showing record arrivals at t1, t2, t6, t7, and multiple at t8 within a 5-second window.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/sliding-10.png) 請謹記以下幾點： + 此範例假設了 5 秒的窗口。5 秒的窗口會隨著時間連續滑動。 + 針對進入窗口的每一列，滑動窗口會發出一個輸出列。應用程式啟動後不久，即使 5 秒的窗口尚未通過，您也會看到查詢發出串流上每筆新記錄的輸出。例如，當記錄出現在第一秒和第二秒的查詢時，串流會發出輸出。稍後，查詢會在 5 秒的窗口中處理記錄。 + 窗口隨著時間滑動。如果串流上的舊記錄落在窗口外，則查詢不會發出輸出，除非串流上也有新記錄落在該 5 秒視窗內。 假設查詢在 t0 開始執行。然後會發生以下情況： 1. t0 時，查詢開始。查詢不會發出輸出（計數值），因為目前沒有記錄。 ![\[Timeline showing a stream starting at t0 with no output initially indicated.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/sliding-t0.png) 1. 在 t1，一個新的記錄出現在串流上，查詢發出計數值 1。 ![\[Timeline showing a stream with a record appearing at time t1, and an arrow pointing to t0.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/sliding-t1.png) 1. 在 t2，另一個記錄出現，查詢發出計數值 2。 ![\[Timeline showing stream events at different time points, with two vertical bars at the end.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/sliding-t2.png) 1. 5 秒的窗口會隨著時間滑動： + 在 t3，滑動窗口 t3 到 t0 + 在 t4 (滑動窗口 t4 到 t0) + 在 t5 滑動窗口 t5-t0 在所有這些時刻，5 秒的窗口具有相同的記錄-沒有新記錄。因此，查詢沒有發出任何輸出。 ![\[Timeline showing stream with multiple time points and colored rectangles representing data windows.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/sliding-t3-4-5.png) 1. 在時間 t6 時，5 秒的窗口為 (t6 到 t1)。查詢在 t6 偵測到一個新的記錄，因此它發出輸出 2。t1 處的記錄不再位於窗口中，並且不會計算在內。 ![\[Timeline showing stream events at different time points with a sliding 5-second window.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/sliding-t6.png) 1. 在時間 t7 時，5 秒的窗口為 t7 到 t2。查詢在 t7 偵測到一個新的記錄，因此它發出輸出 2。t2 處的記錄不再位於 5 秒窗口中，因此不會計算在內。 ![\[Timeline showing stream events and time points from t0 to t7, with a 5-second window highlighted.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/sliding-t7.png) 1. 在時間 t8 時，5 秒的窗口為 t8 到 t3。查詢偵測到三個新記錄，因此會發出記錄計數 5。 ![\[Timeline showing stream events with orange bars representing record counts at different time intervals.\]](http://docs.aws.amazon.com/zh_tw/kinesisanalytics/latest/dev/images/sliding-t8.png) 總之，窗口是一個固定的大小，並隨著時間的推移滑動。新的記錄出現時，查詢會發出輸出。 **注意** 建議您使用不超過一個小時的滑動窗口。如果您使用較長的窗口，在定期系統維護後，應用程式需要更長的時間重新啟動。這是因為必須再次從串流中讀取來源資料。 下列範例查詢使用 `WINDOW` 子句來定義窗口和執行彙總。因為查詢未指定 `GROUP BY`，所以查詢會使用滑動窗口方法來處理串流上的記錄。 ## 範例 1：使用 1 分鐘滑動窗口處理串流 請考慮在入門練習中填入應用程式內串流的示範串流 `SOURCE_SQL_STREAM_001`。以下為其結構描述。 ``` (TICKER_SYMBOL VARCHAR(4), SECTOR varchar(16), CHANGE REAL, PRICE REAL) ``` 假設您希望應用程式使用滑動 1 分鐘窗口來計算彙總。也就是說，對於出現在串流上的每個新記錄，您希望應用程式套用彙總到前 1 分鐘窗口的記錄，以發出輸出。 您可以使用下列時間窗口查詢。查詢會使用 `WINDOW` 子句來定義 1 分鐘的範圍間隔。在 `WINDOW` 子句中，`PARTITION BY`會按滑動窗口內的股票代號值對記錄進行分組。 ``` 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); ``` **若要測試查詢** 1. 按照[入門練習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)設置應用程式。 1. 用上述 `SELECT` 查詢取代應用程式碼中的 `SELECT` 陳述式。產生的應用程式碼如下所示。 ``` 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); ``` ## 範例 2：查詢在滑動視窗上套用彙總 示範串流上的下列查詢，會傳回 10 秒窗口中每個股票代號價格變動百分比的平均值。 ``` 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); ``` **若要測試查詢** 1. 按照[入門練習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)設置應用程式。 1. 用上述 `SELECT` 查詢取代應用程式碼中的 `SELECT` 陳述式。產生的應用程式碼如下所示。 ``` 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); ``` ## 範例 3：查詢相同串流上多個滑動視窗的資料 您可以撰寫查詢以發出輸出，其中每個資料欄值，是使用同一個資料流上定義的不同滑動窗口來計算。 在下面的例子中，查詢發出輸出股票代號，價格，a2 和 a10。它發出輸出給為兩列移動平均線穿過十列移動平均線的股票代碼。`a2` 和 `a10` 欄值衍生自兩列與十列滑動窗口。 ``` 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); ``` 若要針對示範串流測試此查詢，請遵循 [範例 1](#sliding-ex1) 中所述的測試程序。 # 串流資料作業：串流聯結 您可以在應用程式中擁有多個應用程式內串流。您可以撰寫 `JOIN` 查詢來關聯到達這些串流的資料。舉例來說，假設您有以下應用程式內串流。 + **OrderStream**：接收正在下達的庫存訂單。 ``` (orderId SqlType, ticker SqlType, amount SqlType, ROWTIME TimeStamp) ``` + **TradeStream**：接收這些訂單所產生的股票交易。 ``` (tradeId SqlType, orderId SqlType, ticker SqlType, amount SqlType, ticker SqlType, amount SqlType, ROWTIME TimeStamp) ``` 以下是將資料關聯到這些串流的 `JOIN` 查詢範例。 ## 範例 1：報告下訂單後一分鐘內有交易的訂單 在此範例中，您的查詢會同時連接 `OrderStream` 和 `TradeStream`。但是，由於我們只想要訂單後一分鐘下達的交易，因此查詢定義了 1 分鐘的 `TradeStream` 窗口。如需有關窗口查詢的資訊，請參閱[滑動視窗](sliding-window-concepts.md)。 ``` SELECT STREAM ROWTIME, o.orderId, o.ticker, o.amount AS orderAmount, t.amount AS tradeAmount FROM OrderStream AS o JOIN TradeStream OVER (RANGE INTERVAL '1' MINUTE PRECEDING) AS t ON o.orderId = t.orderId; ``` 您可以使用 `WINDOW` 子句明確定義窗口並編寫前面的查詢，如下所示： ``` SELECT STREAM ROWTIME, o.orderId, o.ticker, o.amount AS orderAmount, t.amount AS tradeAmount FROM OrderStream AS o JOIN TradeStream OVER t ON o.orderId = t.orderId WINDOW t AS (RANGE INTERVAL '1' MINUTE PRECEDING) ``` 當您在應用程式碼中包含此查詢時，應用程式碼會持續執行。針對 `OrderStream` 上的每個到達記錄，如果在下訂單後的 1 分鐘窗口內有交易，則應用程序將發出輸出。 在前面的查詢中的連接是一個內部聯接，其中查詢在 `OrderStream` 發出記錄，`TradeStream` 中也有一個相符的記錄（反之亦然）。使用外部連接，您可以創建另一個有趣的情景。假設您想要下單後一分鐘內沒有交易的股票訂單，以及同一窗口其他訂單的交易。此即*外部聯結*的案例。 ``` SELECT STREAM ROWTIME, o.orderId, o.ticker, o.amount AS orderAmount, t.ticker, t.tradeId, t.amount AS tradeAmount, FROM OrderStream AS o LEFT OUTER JOIN TradeStream OVER (RANGE INTERVAL '1' MINUTE PRECEDING) AS t ON o.orderId = t.orderId; ```