範例：使用事件時間戳記的輪轉窗口

PDFRSS

當窗口查詢以非重疊的方式處理每個窗口時，即稱作輪轉窗口。如需詳細資訊，請參閱 輪轉窗口（使用 GROUP BY 彙總）。這個 Amazon Kinesis Data Analytics 範例示範使用事件時間戳記 (即使用者建立，包含在串流資料中的時間戳記) 的輪轉窗口。範例中使用這種方法，而非僅使用 ROWTIME，即 Kinesis Data Analytics 在應用程式收到記錄時所建立的時間戳記。如果您想要根據事件發生的時間 (而非應用程式收到事件的時間) 建立彙總，可在串流資料中使用事件時間戳記。在此範例中，ROWTIME 值會每分鐘觸發彙總，而記錄會依照 ROWTIME 和包含的事件時間兩者進行彙總。

在此範例中，將下列記錄寫入 Amazon Kinesis 串流：此 EVENT_TIME 值設定為過去 5 秒，以模擬處理和傳輸延遲，這些延遲可能會造成事件發生與記錄擷取至 Kinesis Data Analytics 的時間差。

{"EVENT_TIME": "2018-06-13T14:11:05.766191", "TICKER": "TBV", "PRICE": 43.65}
{"EVENT_TIME": "2018-06-13T14:11:05.848967", "TICKER": "AMZN", "PRICE": 35.61}
{"EVENT_TIME": "2018-06-13T14:11:05.931871", "TICKER": "MSFT", "PRICE": 73.48}
{"EVENT_TIME": "2018-06-13T14:11:06.014845", "TICKER": "AMZN", "PRICE": 18.64}
...

然後，您可以在 中建立 Kinesis Data Analytics 應用程式 AWS Management Console，並將 Kinesis 資料串流做為串流來源。探索程序會讀取串流來源上的範例記錄，並推斷含有三個資料欄 (EVENT_TIME、TICKER 和PRICE) 的應用程式內結構描述，如下所示。

您可以將應用程式碼與 MIN 和 MAX 函數搭配使用，以建立資料的窗口化彙總。接著將產生的資料插入另一個應用程式內串流，如下列螢幕擷取畫面所示：

在下列程序中，建立 Kinesis Data Analytics 應用程式，該應用程式會根據事件時間在輪轉窗口中彙總輸入串流的值。

步驟 1：建立 Kinesis Data Stream

建立 Amazon Kinesis 資料串流，並填入紀錄，如下所示：

1. 登入 AWS Management Console ，並在 https：//https://console.aws.amazon.com/kinesis 開啟 Kinesis 主控台。

2. 在導覽窗格中選擇資料串流。

3. 選擇建立 Kinesis 串流，然後建立內含一個碎片之串流。如需詳細資訊，請參閱 Amazon Kinesis Data Streams 開發人員指南中的建立串流。

4. 若要在生產環境中將記錄寫入 Kinesis 資料串流，建議您使用 Kinesis Client Library或 Kinesis Data Streams API。為了簡單起見，這個例子使用下面的 Python 指令碼來生成記錄。執行程式碼以填入範例股票代號記錄。這個簡單的程式碼會持續將隨機股票代號記錄寫入串流。讓指令碼持續執行，以便在稍後的步驟產生應用程式結構描述。

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

步驟 2：建立 Kinesis Data Analytics 應用程式

建立 Kinesis Data Analytics 應用程式，如下所示。

1. 前往 https://console.aws.amazon.com/kinesisanalytics 開啟 Managed Service for Apache Flink 主控台。

2. 選擇建立應用程式，輸入應用程式名稱，然後選擇建立應用程式。

3. 在應用程式詳細資料頁面上，選擇連接串流資料來連接至來源。

4. 在連接至來源頁面，執行下列動作：

   1. 選擇您在上一節建立的串流。

   2. 選擇探索結構描述。等待主控台顯示推斷的結構描述和範例記錄，這些記錄可用來推斷應用程式內串流所建立的結構描述。推斷的結構描述有三個資料欄。

   3. 選擇編輯結構描述。將 EVENT_TIME 欄的欄類型變更為 TIMESTAMP。

   4. 選擇儲存結構描述並更新串流範例。主控台儲存結構描述後，選擇結束。

   5. 選擇儲存並繼續。

5. 在應用程式詳細資訊頁面上，選擇至 SQL 編輯器。若要啟動應用程式，請在出現的對話方塊中選擇是，啟動應用程式。

6. 在 SQL 編輯器中，編寫應用程式碼並驗證結果，如下所示：

   1. 請複製以下應用程式碼，然後貼到編輯器中。

      CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (EVENT_TIME timestamp, TICKER VARCHAR(4), min_price REAL, max_price REAL);
      
      CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
        INSERT INTO "DESTINATION_SQL_STREAM" 
          SELECT STREAM STEP("SOURCE_SQL_STREAM_001".EVENT_TIME BY INTERVAL '60' SECOND),
              TICKER,
               MIN(PRICE) AS MIN_PRICE,
               MAX(PRICE) AS MAX_PRICE
          FROM    "SOURCE_SQL_STREAM_001"
          GROUP BY TICKER, 
                   STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND), 
                   STEP("SOURCE_SQL_STREAM_001".EVENT_TIME BY INTERVAL '60' SECOND);

   2. 選擇 儲存並執行 SQL。

      在即時分析標籤上，您可以查看應用程式建立的所有應用程式內串流，並驗證資料。