範例：轉換 DateTime 值

登入 AWS Management Console 並開啟運動主控台，網址為 https://console.aws.amazon.com/kinesis。

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

選擇建立 Kinesis 串流，然後建立內含一個碎片之串流。

執行下列 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"))

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

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

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

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

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

2. 選擇建立 IAM 角色。

3. 選擇探索結構描述。等待主控台顯示推斷的結構描述，以及用來推斷建立的應用程式內串流結構描述之範例記錄。推斷的結構描述有兩個資料欄。

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

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

6. 選擇儲存並繼續。

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

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

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

   CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
       TICKER VARCHAR(4), 
       event_time TIMESTAMP, 
       five_minutes_before TIMESTAMP, 
       event_unix_timestamp BIGINT,
       event_timestamp_as_char VARCHAR(50),
       event_second INTEGER);
   
   CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
   
   SELECT STREAM 
       TICKER, 
       EVENT_TIME,
       EVENT_TIME - INTERVAL '5' MINUTE,
       UNIX_TIMESTAMP(EVENT_TIME),
       TIMESTAMP_TO_CHAR('yyyy-MM-dd hh:mm:ss', EVENT_TIME),
       EXTRACT(SECOND FROM EVENT_TIME) 
   FROM "SOURCE_SQL_STREAM_001"
   

2. 選擇儲存並執行 SQL。在即時分析標籤上，您可以查看應用程式建立的所有應用程式內串流，並驗證資料。