範例：擷取最常發生的值 (TOP_K_ITEMS_TUMBLING) - 適用於 SQL 應用程式的 Amazon Kinesis Data Analytics 開發人員指南
步驟 1：建立 Kinesis Data Stream步驟 2：建立 Kinesis Data Analytics 應用程式

經過仔細考量，我們決定在兩個步驟中停止 Amazon Kinesis Data Analytics for SQL 應用程式：

1. 從 2025 年 10 月 15 日起，您將無法建立新的 Kinesis Data Analytics for SQL 應用程式。

2. 我們將自 2026 年 1 月 27 日起刪除您的應用程式。您將無法啟動或操作 Amazon Kinesis Data Analytics for SQL 應用程式。從那時起，Amazon Kinesis Data Analytics for SQL 將不再提供支援。如需詳細資訊，請參閱Amazon Kinesis Data Analytics for SQL 應用程式終止。

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

範例：擷取最常發生的值 (TOP_K_ITEMS_TUMBLING)

這個 Amazon Kinesis Data Analytics 範例示範如何使用 TOP_K_ITEMS_TUMBLING 函數擷取輪轉窗口中最常出現的值。如需詳細資訊，請參閱《Amazon Managed Service for Apache Flink SQL 參考資料》中的 TOP_K_ITEMS_TUMBLING 函數。

此 TOP_K_ITEMS_TUMBLING 功能在彙總超過數萬或數十萬個金鑰，且想要減少資源使用量時非常有用。該函數產生與 GROUP BY 和 ORDER BY 子句匯總相同的結果。

在此範例中，將下列記錄寫入 Amazon Kinesis 資料串流：

{"TICKER": "TBV"}
{"TICKER": "INTC"}
{"TICKER": "MSFT"}
{"TICKER": "AMZN"}
...

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

顯示應用程式內結構描述的主控台螢幕擷取畫面，其中包含股票代碼欄。

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

顯示應用程式內串流生成資料的主控台螢幕擷取畫面。

在下列程序中，建立 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. 選擇儲存結構描述並更新串流範例。主控台儲存結構描述後，選擇結束。

    4. 選擇儲存並繼續。

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

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

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

      CREATE OR REPLACE STREAM DESTINATION_SQL_STREAM (
  "TICKER" VARCHAR(4), 
  "MOST_FREQUENT_VALUES" BIGINT
);

CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
    INSERT INTO "DESTINATION_SQL_STREAM"
    SELECT STREAM * 
        FROM TABLE (TOP_K_ITEMS_TUMBLING(
            CURSOR(SELECT STREAM * FROM "SOURCE_SQL_STREAM_001"),
            'TICKER',         -- name of column in single quotes
            5,                       -- number of the most frequently occurring values
            60                       -- tumbling window size in seconds
            )
        );

    2. 選擇 儲存並執行 SQL。

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