例如：检索最常出现的值 (TOP_K_ITEMS_TUMBLING) - Amazon Kinesis Data Analytics 开发人员指南
第 1 步：创建 Kinesis Data Streams第 2 步：创建 Kinesis Data Analytics 应用程序
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅中国的 Amazon Web Services 服务入门。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

例如：检索最常出现的值 (TOP_K_ITEMS_TUMBLING)

此 Amazon Kinesis Data Analytics 示例演示如何使用TOP_K_ITEMS_TUMBLING函数用于在滚动窗口中检索最常出现的值。有关更多信息，请参阅 。TOP_K_ITEMS_TUMBLING功能中的Amazon Kinesis Data Analytics SQL 参考.

在对数万或数十万个密钥进行聚合时，如果您希望降低资源占用，则 TOP_K_ITEMS_TUMBLING 函数很有用。该函数生成的结果与使用 GROUP BY 和 ORDER BY 子句进行聚合一样。

在本示例中，您将以下记录写入到 Amazon Kinesis 数据流中：

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

然后 Kinesis Data Analytics 在Amazon Web Services Management Console，将 Kinesis 数据流用作流式源。发现过程读取流式传输源上的示例记录，并推断出具有一个列 (TICKER) 的应用程序内部架构，如下所示：

控制台屏幕截图，显示具有股票代码列的应用程序内部架构。

您使用应用程序代码以及 TOP_K_VALUES_TUMBLING 函数以创建数据的窗口式聚合。然后，将结果数据插入另一个应用程序内部流，如下面的屏幕截图所示：

控制台屏幕截图，显示应用程序内部流中的结果数据。

在以下过程中，您将创建 Kinesis Data Analytics 应用程序，用于在输入流中检索最常出现的值。

第 1 步：创建 Kinesis Data Streams

按如下方式创建 Amazon Kinesis 数据流并填充记录：

  1. 登录到 Amazon Web Services Management Console，然后通过以下网址打开 Kinesis 控制台：https://console.aws.amazon.com/kinesisvideo/home。

  2. 在导航窗格中，选择 Data Streams (数据流)。

  3. 选择 Create Kinesis stream (创建 Kinesis 流)，然后创建具有一个分片的流。有关更多信息，请参阅 。创建流中的Amazon Kinesis Data Streams 开发人员指南.

  4. 要在生产环境中将记录写入到 Kinesis 数据流，我们建议您使用 Kinesis 客户端库或 Kinesis 数据流 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. 打开 Kinesis Data Analytics 控制台https://console.aws.amazon.com/kinesisanalytics.

  2. 选择 Create application (创建应用程序)，键入应用程序名称，然后选择 Create application (创建应用程序)。

  3. 在应用程序详细信息页面上，选择 Connect streaming data (连接流数据)，以连接到源。

  4. 在 Connect to source (连接到源) 页面上，执行以下操作：

    1. 选择在上一部分中创建的流。

    2. 选择 Discover Schema (发现架构)。等待控制台显示推断的架构和为创建的应用程序内部流推断架构所使用的示例记录。推断的架构包含一列。

    3. 选择 Save schema and update stream samples。在控制台保存架构后，选择 Exit (退出)。

    4. 选择 Save and continue。

  5. 在应用程序详细信息页面上，选择 Go to SQL editor (转到 SQL编辑器)。要启动应用程序，请在显示的对话框中选择 Yes, start application (是，启动应用程序)。

  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. 选择 Save and run SQL。

      在 Real-time analytics (实时分析) 选项卡上，可以查看应用程序已创建的所有应用程序内部流并验证数据。