本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
示例:在 Python 中创建滑动窗口
在本练习中,您创建一个使用滑动窗口聚合数据的 Python Kinesis Data Analytics 应用程序。在 Flink 中,该功能状态。要禁用终止保护,请使用以下内容:
sink.producer.aggregation-enabled' = 'false'
要为本练习设置所需的先决条件,请先完成入门 (Python)练习。
本主题包含下列部分:
创建相关资源
在为本练习创建 Kinesis Data Analytics 应用程序之前,请创建以下相关资源:
两个 Kinesis 数据流 (
ExampleInputStream和ExampleOutputStream)Amazon S3 存储桶 (
ka-app-code-)<username>
您可以使用控制台创建 Kinesis 流和 Amazon S3 存储桶。有关创建这些资源的说明,请参阅以下主题:
创建和更新数据流中的Amazon Kinesis Data Streams 开发者指南. 将数据流命名为
ExampleInputStream和ExampleOutputStream。如何创建 S3 存储桶?中的Amazon Storage Service. 附加您的登录名,以便为 Amazon S3 存储桶指定全局唯一的名称,例如
ka-app-code-.<username>
将示例记录写入输入流
在本节中,您使用 Python 脚本将示例记录写入流,以供应用程序处理。
此部分需要 Amazon SDK for Python (Boto)
本节中的 Python 脚本使用Amazon CLI. 你必须配置你的Amazon CLI使用您的账户凭证和默认区域。配置您的Amazon CLI,输入以下内容:
aws configure
-
使用以下内容创建名为
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')) -
运行
stock.py脚本:$ python stock.py在完成本教程的其余部分时,请将脚本保持运行状态。
下载并检查应用程序代码
该示例的 Python 应用程序代码可从 GitHub. 要下载应用程序代码,请执行以下操作:
如果尚未安装 Git 客户端,请安装它。有关更多信息,请参阅安装 Git
。 使用以下命令克隆远程存储库:
git clone https://github.com/aws-samples/>amazon-kinesis-data-analytics-java-examples导航到
amazon-kinesis-data-analytics-java-examples/python/SlidingWindow目录。
应用程序代码位于 sliding-windows.py 文件中。请注意有关应用程序代码的以下信息:
应用程序使用 Kinesis 表源从源流中进行读取。以下片段调用了
create_table函数来创建 Kinesis 表源:table_env.execute_sql( create_table(input_table_name, input_stream, input_region, stream_initpos) )这些区域有:
create_table函数使用 SQL 命令创建由流式处理源支持的表:def create_table(table_name, stream_name, region, stream_initpos): return """ CREATE TABLE {0} ( ticker VARCHAR(6), price DOUBLE, event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND ) PARTITIONED BY (ticker) WITH ('connector' = 'kinesis','stream' = '{1}', 'aws.region' = '{2}', 'scan.stream.initpos' = '{3}', 'sink.partitioner-field-delimiter' = ';', 'sink.producer.collection-max-count' = '100', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' ) """.format( table_name, stream_name, region, stream_initpos )应用程序使用
Slide运算符聚合指定滑动窗口内的记录,并将聚合记录作为表对象返回:sliding_window_table = ( input_table.window(Slide.over("10.seconds").every("5.seconds") .on("event_time") .alias("ten_second_window") ) .group_by("ticker, ten_second_window") .select("ticker, price.min as price, ten_second_window.end as event_time") )该应用程序使用 Kinesis Flink 连接器,来自flink-sql-connector-kinesis_2.12
文件。
压缩并上传 Apache Flink 流式处理 Python 代码
在本节中,您将应用程序代码上传到在一节中创建的 Amazon S3 存储桶创建相关资源部分。
使用您首选的压缩应用程序来压缩
streaming-file-sink.py和flink-sql-connector-kinesis_2.12-1.13.2.jar文件。命名存档名称myapp.zip.-
在 Amazon S3 控制台中,选择ka-app-code-
<username>存储桶,然后选择上传. -
在选择文件步骤中,选择添加文件。导航到您在上一步中创建的
myapp.zip文件。 您无需更改该对象的任何设置,因此,请选择 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.132。
将版本下拉菜单保留为Apache Flink 1.13.2(推荐版本).
-
-
对于访问权限,请选择 Create / update IAM role (创建/更新 IAM 角色)
kinesis-analytics-MyApplication-us-west-2。 -
选择 Create application(创建应用程序)。
在使用控制台创建 Kinesis Data Analts 应用程序时,您可以选择为应用程序创建一个 IAM 角色和策略。您的应用程序使用此角色和策略访问其从属资源。这些 IAM 资源是使用您的应用程序名称和区域命名的,如下所示:
-
策略:
kinesis-analytics-service-MyApplication-us-west-2 -
角色:
kinesis-analytics-MyApplication-us-west-2
配置应用程序
-
在存储库的MyApplication页面上,选择配置.
-
在 Configure application (配置应用程序) 页面上,提供 Code location (代码位置):
-
适用于Amazon S3 存储桶输入
ka-app-code-.<username> -
适用于Amazon S3 对象的路径输入
myapp.zip.
-
-
在 Access to application resources (对应用程序的访问权限) 下,对于 Access permissions (访问权限),选择 Create / update IAM role (创建/更新 IAM 角色)
kinesis-analytics-MyApplication-us-west-2。 -
UNDER属性,选择添加组. 适用于Group ID (组 ID)输入
consumer.config.0. -
输入以下应用程序属性和值:
密钥 值 input.stream.nameExampleInputStreamaws.regionus-west-2flink.stream.initposLATEST选择Save(保存)。
UNDER属性,选择添加组。适用于Group ID (组 ID)输入
producer.config.0.输入以下应用程序属性和值:
密钥 值 output.stream.nameExampleOutputStreamaws.regionus-west-2shard.count1UNDER属性,选择添加组。适用于Group ID (组 ID)输入
kinesis.analytics.flink.run.options. 这个特殊的属性组告诉你的应用程序在哪里可以找到它的代码资源。有关更多信息,请参阅 指定代码文件。输入以下应用程序属性和值:
密钥 值 pythonsliding-windows.pyjarfileamazon-kinesis-connector-flink-2.0.0.jar-
在 Monitoring (监控) 下,确保 Monitoring metrics level (监控指标级别) 设置为 Application (应用程序)。
-
适用于CloudWatch 记录,选择启用”复选框。
-
选择 Update(更新)。
当你选择启用 CloudWatch 日志记录时,Kinesis Data Analytics 将为您创建日志组和日志流。这些资源的名称如下所示:
-
日志组:
/aws/kinesis-analytics/MyApplication -
日志流:
kinesis-analytics-log-stream
该日志流用于监控应用程序。这与应用程序用于发送结果的日志流不同。
编辑 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>/myapp.zip" ] }, { "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" }] }
运行应用程序
通过运行应用程序,打开 Apache Flink 控制面板,然后选择所需的 Flink 作业,可以查看 Flink 作业图。
你可以在上查看 Kinesis Data Analytics 指标 CloudWatch 控制台,以验证应用程序是否正常工作。
清理 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 Streams然后确认删除。
在Kinesis Streams页面上,选择ExampleOutputStream,选择操作,选择Delete,然后确认删除。
删除您的 Amazon S3 对象和存储桶
通过以下网址打开 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(角色)。
选择kinesines-anticticMyApplication-
<your-region>角色。选择 Delete role (删除角色),然后确认删除。
删除日期和时间 CloudWatch 资源
打开 CloudWatch 控制台https://console.aws.amazon.com/cloudwatch/
. 在导航栏中,选择 Logs (日志)。
选择/aws/kinesis-analytictics/MyApplication日志组。
选择 Delete Log Group (删除日志组),然后确认删除。