示例：转换 DateTime 值

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

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

选择 Create Kinesis stream (创建 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'))

打开 Kinesis Data Analytics 控制台，网址为 https://console.aws.amazon.com/kinesisanalytics。

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

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

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

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

2. 选择以创建 IAM 角色。

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

4. 选择 Edit Schema (编辑架构)。将 EVENT_TIME 列的 Column type (列类型) 更改为 TIMESTAMP。

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

6. 选择 Save and continue。

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

在 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. 选择 Save and run SQL。在 Real-time analytics (实时分析) 选项卡上，可以查看应用程序已创建的所有应用程序内部流并验证数据。