示例：转换 DateTime 值

登录 AWS Management Console 并打开 Kinesis 控制台，网址为 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"))

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

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

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

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

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