本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
使用 Apache Beam 创建应用程序
在本练习中,您将创建一个使用 Apache Beam
注意
要为本练习设置所需的先决条件,请先完成入门指南 (DataStream API)练习。
本主题包含下列部分:
创建相关资源
在为本练习创建 Kinesis Data Analytics 应用程序之前,您需要创建以下依赖资源:
两个 Kinesis 数据流(
ExampleInputStream和ExampleOutputStream)用于存储应用程序代码的 Amazon S3 存储桶 () 存储桶 (
ka-app-code-)<username>
您可以使用控制台创建 Kinesis Streams 和 Amazon S3 存储桶。有关创建这些资源的说明,请参阅以下主题:
Amazon Kinesis Data Streams? 将数据流命名为
ExampleInputStream和ExampleOutputStream。如何创建 S3 存储桶? Amazon Simple Storage 用户指南。通过附加您的登录名,为 Amazon S3 存储桶指定一个全球唯一的名称,例如
ka-app-code-。<username>
将示例记录写入输入流
在本节中,您将使用 Python 脚本将随机字符串写入流以供应用程序处理。
注意
此部分需要 Amazon SDK for Python (Boto)
-
使用以下内容创建名为
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在完成本教程的其余部分时,请将脚本保持运行状态。
下载并检查应用程序代码
此示例的 Java 应用程序代码可从以下网址获得 GitHub。要下载应用程序代码,请执行以下操作:
如果尚未安装 Git 客户端,请安装它。有关更多信息,请参阅安装 Git
。 使用以下命令克隆远程存储库:
git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-java-examples导航到
amazon-kinesis-data-analytics-java-examples/Beam目录。
应用程序代码位于 BasicBeamStreamingJob.java 文件中。请注意有关应用程序代码的以下信息:
该应用程序使用 Apache Beam 通过调用名ParDo
为的自定义转换函数来处理传入的记录 PingPongFn。调用该
PingPongFn函数的代码如下所示:.apply("Pong transform", ParDo.of(new PingPongFn())使用 Apache Beam 的 Kinesis Data Analytics 应用程序需要以下组件。如果您不在您的
pom.xml中包含这些组件和版本,则您的应用程序会从环境依赖项中加载错误的版本,并且由于版本不匹配,您的应用程序会在运行时崩溃。<jackson.version>2.10.2</jackson.version> ... <dependency> <groupId>com.fasterxml.jackson.module</groupId> <artifactId>jackson-module-jaxb-annotations</artifactId> <version>2.10.2</version> </dependency>PingPongFn转换函数将输入数据传递到输出流中,除非输入数据为 ping,在这种情况下,它会向输出流发出字符串 pong\ n。转换函数的代码如下:
private static class PingPongFn extends DoFn<KinesisRecord, byte[]> { private static final Logger LOG = LoggerFactory.getLogger(PingPongFn.class); @ProcessElement public void processElement(ProcessContext c) { String content = new String(c.element().getDataAsBytes(), StandardCharsets.UTF_8); if (content.trim().equalsIgnoreCase("ping")) { LOG.info("Ponged!"); c.output("pong\n".getBytes(StandardCharsets.UTF_8)); } else { LOG.info("No action for: " + content); c.output(c.element().getDataAsBytes()); } } }
编译应用程序代码
要编译应用程序,请执行以下操作:
如果还没有 Java 和 Maven,请安装它们。有关更多信息,请参阅入门指南 (DataStream API)教程中的先决条件。
使用以下命令编译应用程序:
mvn package -Dflink.version=1.15.3 -Dflink.version.minor=1.8注意
提供的源代码依赖于 Java 11 中的库。
编译应用程序将创建应用程序 JAR 文件 (target/basic-beam-app-1.0.jar)。
上传 Apache Flink 流式处理 Java 代码
在本部分中,您将应用程序代码上传到您在该创建相关资源部分创建的 Amazon S3 存储桶。
-
在 Amazon S3 控制台中,选择 ka-app-code-
<username>存储桶,然后选择上传。 -
在选择文件步骤中,选择添加文件。导航到您在上一步中创建的
basic-beam-app-1.0.jar文件。 您无需更改该对象的任何设置,因此,请选择 Upload (上传)。
您的应用程序代码现存储在 Amazon S3 存储桶中,应用程序可以在此存储桶中进行访问。
创建并运行 Kinesis Data Analytics 应用程序
按照以下步骤,使用控制台创建、配置、更新和运行应用程序。
创建 应用程序
打开 Kinesis Data Analytics 控制台,网址为 https://console.aws.amazon.com/kinesisanalytics
。 -
在 Kinesis Data Analytics 仪表板上,选择创建分析应用程序。
-
在 Kinesis Analytics – 创建应用程序页面上,提供应用程序详细信息,如下所示:
-
对于 Application name (应用程序名称),输入
MyApplication。 -
对于 Runtime (运行时),请选择 Apache Flink。
注意
Kinesis Data Analytics 使用 Apache Flink 版本 1.15.2。
将版本下拉列表保留为 Apache Flink 版本 1.15.2(推荐版本)。
-
-
对于访问权限,请选择 Create / update IAM role (创建/更新 IAM 角色)
kinesis-analytics-MyApplication-us-west-2。 -
选择创建应用程序。
注意
当您使用控制台创建 Kinesis Data Analytics 应用程序时,您可以选择为应用程序创建 IAM 角色和策略。您的应用程序使用此角色和策略访问其从属资源。这些 IAM 资源使用您的应用程序名称和区域命名,如下所示:
-
策略:
kinesis-analytics-service-MyApplication-us-west-2 -
角色:
kinesis-analytics-MyApplication-us-west-2
编辑 IAM 策略
编辑 IAM 策略以添加访问 Kinesis 数据流的权限。
通过以下网址打开 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>/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页面上,选择配置。
-
在 Configure application (配置应用程序) 页面上,提供 Code location (代码位置):
-
对于 Amazon S3 存储桶,输入
ka-app-code-。<username> -
在 Amazon S3 对象的路径中,输入
basic-beam-app-1.0.jar。
-
-
在 Access to application resources (对应用程序的访问权限) 下,对于 Access permissions (访问权限),选择 Create / update IAM role (创建/更新 IAM 角色)
kinesis-analytics-MyApplication-us-west-2。 -
输入以下信息:
组 ID 键 值 BeamApplicationPropertiesInputStreamNameExampleInputStreamBeamApplicationPropertiesOutputStreamNameExampleOutputStreamBeamApplicationPropertiesAwsRegionus-west-2 -
在 Monitoring (监控) 下,确保 Monitoring metrics level (监控指标级别) 设置为 Application (应用程序)。
-
要进行CloudWatch 记录,请选中 “启用” 复选框。
-
选择更新。
注意
当您选择启用 CloudWatch 日志记录时,Kinesis Data Analytics 会为您创建日志组和日志流。这些资源的名称如下所示:
-
日志组:
/aws/kinesis-analytics/MyApplication -
日志流:
kinesis-analytics-log-stream
该日志流用于监控应用程序。这与应用程序用于发送结果的日志流不同。
运行应用程序
可以通过运行应用程序、打开 Apache Flink 仪表板并选择所需的 Flink 作业来查看 Flink 任务图。
您可以在 CloudWatch 控制台上查看 Kinesis Data Analytics 指标,以验证应用程序是否正常运行。
清理 Amazon 资源
本节包含清理在滚动窗口教程中创建的 Amazon 资源的过程。
本主题包含下列部分:
删除 Kinesis Data Analytics
打开 Kinesis Data Analytics 控制台,网址为 https://console.aws.amazon.com/kinesisanalytics
。 在 Kinesis Data Analytics 面板中,选择MyApplication。
在应用程序的页面中,选择 Delete (删除),然后确认删除。
删除 Kinesis Data Streams
打开 Kinesis 控制台,网址为:https://console.aws.amazon.com/kinesis
。 在 Kinesis Data Streams 面板中,选择ExampleInputStream。
在ExampleInputStream页面中,选择 “删除 Kinesis Stream”,然后确认删除。
在 Kinesis stream s 页面中 ExampleOutputStream,选择,选择 “操作”,选择 “删除”,然后确认删除。
删除您的Amazon S3 对象和存储桶
通过以下网址打开 Simple Storage Service(Amazon S3)控制台:https://console.aws.amazon.com/s3/
。 选择 ka-app-code-
存储桶。 <username>选择 Delete (删除),然后输入存储桶名称以确认删除。
删除您的 IAM 资源
通过以下网址打开 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 (删除角色),然后确认删除。
删除您的 CloudWatch 资源
通过 https://console.aws.amazon.com/cloudwatch/
打开 CloudWatch 主机。 在导航栏中,选择 Logs (日志)。
选择 /aws/kinesis-analytics/MyApplication 日志组。
选择 Delete Log Group (删除日志组),然后确认删除。
后续步骤
现在,您已经创建并运行了使用 Apache Beam 转换数据的基本 Kinesis Data Analytics 应用程序,请参阅以下应用程序,了解更高级的 Kinesis Data Analytics 解决方案的示例。
Beam on Kinesis Data Analytics 直播研讨
会:在本次研讨会中,我们将探讨一个端到端的示例,该示例将批处理和串流方面结合在一个统一的 Apache Beam 流水线中。