使用 Apache Beam 创建应用程序

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

import json
import boto3
import random

kinesis = boto3.client('kinesis')

while True:
        data = random.choice(['ping', 'telnet', 'ftp', 'tracert', 'netstat'])
        print(data)
        kinesis.put_record(
                StreamName="ExampleInputStream",
                Data=data,
                PartitionKey="partitionkey")

运行 ping.py 脚本：

$ python ping.py

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

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

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

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

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

如果还没有 Java 和 Maven，请安装它们。有关更多信息，请参阅教程：开始使用适用于 Apache Flink 的托管服务中的 DataStream API教程中的完成必需的先决条件。

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

mvn package -Dflink.version=1.15.2 -Dflink.version.minor=1.8

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

在选择文件步骤中，选择添加文件。导航到您在上一步中创建的 basic-beam-app-1.0.jar 文件。

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

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

在 Managed Service for Apache Flink 控制面板上，选择创建分析应用程序。

在Managed Service for Apache Flink - 创建应用程序页面上，提供应用程序详细信息，如下所示：

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

选择创建应用程序。

使用 https://console.aws.amazon.com/iam/ 打开 IAM 控制台。

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

在 摘要 页面上，选择 编辑策略。选择 JSON 选项卡。

将以下策略示例中突出显示的部分添加到策略中。请将占位符账户 IDs (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>/basic-beam-app-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": "ReadInputStream",
            "Effect": "Allow",
            "Action": "kinesis:*",
            "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"
        }
    ]
}

在MyApplication页面上，选择配置。

在 配置应用程序 页面上，提供 代码位置：

在 对应用程序的访问权限 下，对于 访问权限，选择 创建/更新 IAM 角色 kinesis-analytics-MyApplication-us-west-2。

输入以下信息：

在 监控 下，确保 监控指标级别 设置为 应用程序。

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

选择更新。

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

在适用于 Apache Flink 的托管服务面板中，选择。MyApplication

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

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

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

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

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

打开 Amazon S3 控制台，网址为 https://console.aws.amazon.com/s3/。

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

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

使用 https://console.aws.amazon.com/iam/ 打开 IAM 控制台。

在导航栏中，选择策略。

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

选择 kinesis-analytics-service--us-MyApplication west-2 策略。

选择 策略操作，然后选择 删除。

在导航栏中，选择 角色。

选择 k inesis-analytics-west-2 角色MyApplication。

选择 删除角色，然后确认删除。

打开 CloudWatch 控制台，网址为https://console.aws.amazon.com/cloudwatch/。

在导航栏中，选择 日志。

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

选择 删除日志组，然后确认删除。