示例：交错窗口

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

在导航窗格中，选择 数据流。

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

要在生产环境中将记录写入到 Kinesis 数据流，我们建议您使用 Kinesis 创建器库或 Kinesis 数据流 API。为简单起见，此示例使用以下 Python 脚本以便生成记录。运行此代码以填充示例股票代码记录。这段简单代码在一分钟时间中连续地将一组六个记录与相同的随机 EVENT_TIME 和股票代码符号一起写入流中。让脚本保持运行，以便可以在后面的步骤中生成应用程序架构。

 
import datetime
import json
import random
import time
import boto3

STREAM_NAME = "ExampleInputStream"


def get_data():
    event_time = datetime.datetime.utcnow() - datetime.timedelta(seconds=10)
    return {
        "EVENT_TIME": event_time.isoformat(),
        "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
    }


def generate(stream_name, kinesis_client):
    while True:
        data = get_data()
        # Send six records, ten seconds apart, with the same event time and ticker
        for _ in range(6):
            print(data)
            kinesis_client.put_record(
                StreamName=stream_name,
                Data=json.dumps(data),
                PartitionKey="partitionkey",
            )
            time.sleep(10)


if __name__ == "__main__":
    generate(STREAM_NAME, boto3.client("kinesis"))

打开适用于 Apache Flink 的托管服务控制台，网址为 https://console.aws.amazon.com/kinesisanalytics。

选择 创建应用程序，键入应用程序名称，然后选择 创建应用程序。

在应用程序详细信息页面上，选择 连接流数据，以连接到源。

在 连接到源 页面上，执行以下操作：

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

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

3. 选择 编辑架构。将 EVENT_TIME 列的 列类型 更改为 TIMESTAMP。

4. 选择 保存架构并更新流示例。在控制台保存架构后，选择 退出。

5. 选择 保存并继续。

在应用程序详细信息页面上，选择 转到 SQL编辑器。要启动应用程序，请在显示的对话框中选择 是，启动应用程序。

在 SQL 编辑器中编写应用程序代码并确认结果如下所示：

1. 复制下面的应用程序代码并将其粘贴到编辑器中。

   CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
       event_time TIMESTAMP,
       ticker_symbol    VARCHAR(4),
       ticker_count     INTEGER);
   
   CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
     INSERT INTO "DESTINATION_SQL_STREAM" 
       SELECT STREAM 
           EVENT_TIME, 
           TICKER,
           COUNT(TICKER) AS ticker_count
       FROM "SOURCE_SQL_STREAM_001"
       WINDOWED BY STAGGER (
               PARTITION BY TICKER, EVENT_TIME RANGE INTERVAL '1' MINUTE);

2. 选择 保存并运行 SQL。

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