示例：将 EFO 消费者与 Kinesis 数据流一起使用

使用以下内容创建名为 stock.py 的文件：

 
    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', region_name='us-west-2'))

运行 stock.py 脚本：

$ python stock.py

在完成本教程的其余部分时，请将脚本保持运行状态。

如果尚未安装 Git 客户端，请安装它。有关更多信息，请参阅安装 Git。

使用以下命令克隆远程存储库：

git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-java-examples

导航到 amazon-kinesis-data-analytics-java-examples/EfoConsumer目录。

如果还没有 Java 和 Maven，请安装它们。有关更多信息，请参阅入门指南 (DataStream API)教程中的先决条件。

使用以下命令编译应用程序：

mvn package -Dflink.version=1.15.3

在 Amazon S3 控制台中，选择 ka-app-code- <username>存储桶，然后选择上传。

在选择文件步骤中，选择添加文件。导航到您在上一步中创建的 aws-kinesis-analytics-java-apps-1.0.jar 文件。

您无需更改该对象的任何设置，因此，请选择 Upload (上传)。

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

在 Kinesis Data Analytics 仪表板上，选择创建分析应用程序。

在 Kinesis Analytics – 创建应用程序页面上，提供应用程序详细信息，如下所示：

对于访问权限，请选择 Create / update IAM role (创建/更新 IAM 角色) kinesis-analytics-MyApplication-us-west-2。

选择创建应用程序。

通过以下网址打开 IAM 控制台：https://console.aws.amazon.com/iam/。

选择策略。选择控制台在上一部分中为您创建的 kinesis-analytics-service-MyApplication-us-west-2 策略。

在 Summary (摘要) 页面上，选择 Edit policy (编辑策略)。请选择 JSON 选项卡。

将以下策略示例中突出显示的部分添加到策略中。将示例账户 ID (012345678901) 替换为您的账户 ID。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ReadCode",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "logs:DescribeLogGroups",
                "s3:GetObjectVersion"
            ],
            "Resource": [
                "arn:aws:logs:us-west-2:012345678901:log-group:*",
                "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
            ]
        },
        {
            "Sid": "DescribeLogStreams",
            "Effect": "Allow",
            "Action": "logs:DescribeLogStreams",
            "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
        },
        {
            "Sid": "PutLogEvents",
            "Effect": "Allow",
            "Action": "logs:PutLogEvents",
            "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
        },
        {
            "Sid": "ListCloudwatchLogGroups",
            "Effect": "Allow",
            "Action": [
                "logs:DescribeLogGroups"
            ],
            "Resource": [
                "arn:aws:logs:us-west-2:012345678901:log-group:*"
            ]
        },
        {
            "Sid": "AllStreams",
            "Effect": "Allow",
            "Action": [
                "kinesis:ListShards",
                "kinesis:ListStreamConsumers",
                "kinesis:DescribeStreamSummary"
            ],
            "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/*"
        },
        {
            "Sid": "Stream",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:RegisterStreamConsumer",
                "kinesis:DeregisterStreamConsumer"
            ],
            "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
        },
        {
            "Sid": "WriteOutputStream",
            "Effect": "Allow",
            "Action": "kinesis:*",
            "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
        },
        {
            "Sid": "Consumer",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStreamConsumer",
                "kinesis:SubscribeToShard"
            ],
            "Resource": [
                "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app",
                "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app:*"
            ]
        }
    ]
}

在MyApplication页面上，选择配置。

在 Configure application (配置应用程序) 页面上，提供 Code location (代码位置)：

在 Access to application resources (对应用程序的访问权限) 下，对于 Access permissions (访问权限)，选择 Create / update IAM role (创建/更新 IAM 角色) kinesis-analytics-MyApplication-us-west-2。

在 “属性” 下，选择 “创建群组”。

输入以下应用程序属性和值：

在 “属性” 下，选择 “创建群组”。

输入以下应用程序属性和值：

在 Monitoring (监控) 下，确保 Monitoring metrics level (监控指标级别) 设置为 Application (应用程序)。

要进行CloudWatch 记录，请选中 “启用” 复选框。

选择更新。

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

在 Kinesis Data Analytics 面板中，选择MyApplication。

在应用程序的页面中，选择 Delete (删除)，然后确认删除。

打开 Kinesis 控制台，网址为：https://console.aws.amazon.com/kinesis。

在 Kinesis Data Streams 面板中，选择ExampleInputStream。

在ExampleInputStream页面中，选择 “删除 Kinesis Stream”，然后确认删除。

在 Kinesis stream s 页面中 ExampleOutputStream，选择，选择操作，选择删除，然后确认删除。

通过以下网址打开 Simple Storage Service（Amazon S3）控制台：https://console.aws.amazon.com/s3/。

选择 ka-app-code- 存储桶。 <username>

选择 Delete (删除)，然后输入存储桶名称以确认删除。

通过以下网址打开 IAM 控制台：https://console.aws.amazon.com/iam/。

在导航栏中，选择策略。

在筛选条件控件中，输入 kinesis。

选择 kinesis-analytics-service-MyApplication- <your-region>策略。

选择 Policy Actions (策略操作)，然后选择 Delete (删除)。

在导航栏中，选择 Roles（角色）。

选择 k inesis-analyticsMyApplication- <your-region>角色。

选择 Delete role (删除角色)，然后确认删除。

通过 https://console.aws.amazon.com/cloudwatch/ 打开 CloudWatch 主机。

在导航栏中，选择 Logs (日志)。

选择 /aws/kinesis-analytics/MyApplication 日志组。

选择 Delete Log Group (删除日志组)，然后确认删除。