示例：根据正则表达式分析日志字符串 (REGEX_LOG_PARSE 函数)

登录 Amazon Web Services Management Console 并在 /kinesis 上打开 Kinesis 控制台。https://console.aws.amazon.com

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

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

运行以下 Python 代码以便填充示例日志记录。这段简单代码不断地将同一日志记录写入到流中。

 
import json
import boto3

STREAM_NAME = "ExampleInputStream"


def get_data():
    return {
        "LOGENTRY": "203.0.113.24 - - [25/Mar/2018:15:25:37 -0700] "
        '"GET /index.php HTTP/1.1" 200 125 "-" '
        '"Mozilla/5.0 [en] Gecko/20100101 Firefox/52.0"'
    }


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"))

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

选择 Create application，并指定应用程序名称。

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

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

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

2. 选择创建 IAM 角色的选项。

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

4. 选择 保存并继续。

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

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

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

   CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (logentry VARCHAR(24), match1 VARCHAR(24), match2 VARCHAR(24));
   
   CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
       SELECT STREAM T.LOGENTRY, T.REC.COLUMN1, T.REC.COLUMN2
       FROM 
            (SELECT STREAM LOGENTRY,
                REGEX_LOG_PARSE(LOGENTRY, '(\w.+) (\d.+) (\w.+) (\w.+)') AS REC
                FROM SOURCE_SQL_STREAM_001) AS T;

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