创建并运行面向 Python 应用程序的 Managed Service for Apache Flink

创建数据流 (Amazon CLI)

1. 要创建第一个流 (ExampleInputStream)，请使用以下 Amazon Kinesis create-streamAmazon CLI命令。

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

2. 要创建应用程序用来写入输出的第二个流，请运行同一命令（将流名称更改为 ExampleOutputStream）。

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

1. 使用以下内容创建名为 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'))

2. 运行 stock.py 脚本：

   $ python stock.py

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

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

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

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

3. 导航到 amazon-kinesis-data-analytics-java-examples/python/GettingStarted 目录。

1. 在本地计算机上创建具有所需依赖项的独立 Python 环境（conda 或类似环境）。

2. 记下该环境site_packages文件夹中的初始软件包列表。

3. 您的应用程序所需的pip-install所有依赖项。

4. 记下在上述步骤 3 之后添加到该site_packages文件夹的软件包。这些是您需要包含在软件包中的文件夹（在my_deps文件夹下），其组织方式如上所示。这将允许您捕获步骤 2 和步骤 3 之间的软件包差异，从而为您的应用程序确定正确的软件包依赖关系。

5. 在kinesis.analytics.flink.run.options属性组中my_deps/作为pyFiles属性的参数提供，如下jarfiles属性所述。Flink 还允许您使用 add_python_file 函数指定 Python 依赖关系，但请务必记住，只需要指定其中一个，而不是两者兼而有之。

要使用控制台上传应用程序代码，请执行以下操作：

1. 使用你首选的压缩应用程序压缩getting-started.py和 https://mvnrepository.com/artifact/org.apache.flink/flink- sql-connector-kinesis _2.12/1. 15.2 文件。为存档myapp.zip命名。如果您在存档中包含外部文件夹，则必须将其包含在路径中，并在配置文件中添加代码：GettingStarted/getting-started.py。

2. 通过 https://console.aws.amazon.com/s3/ 打开 Amazon S3 控制台。

3. 选择 创建存储桶 。

4. 在 存储桶名称 字段中输入 ka-app-code-<username>。将后缀（如您的用户名）添加到存储桶名称，以使其具有全局唯一性。选择 下一步。

5. 在配置选项步骤中，让设置保持原样，然后选择下一步。

6. 在设置权限步骤中，让设置保持原样，然后选择下一步。

7. 选择 创建存储桶 。

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

9. 在选择文件步骤中，选择添加文件。导航到您在上一步中创建的 myapp.zip 文件。选择 下一步。

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

要使用Amazon CLI上传应用程序代码：

1. 使用你首选的压缩应用程序压缩streaming-file-sink.py和 https://mvnrepository.com/artifact/org.apache.flink/flink- sql-connector-kinesis _2.12/1. 15.2 文件。

   注意

   请勿使用 Finder (macOS) 或 Windows 资源管理器 (Windows) 中的压缩功能来创建 myapp.zip 存档。这可能导致应用程序代码无效。

2. 使用你首选的压缩应用程序来压缩getting-started.py和 https://mvnrepository.com/artifact/org.apache.flink/ flink-sql-connector-kinesis /1.15.2 文件。为存档myapp.zip命名。如果您在存档中包含外部文件夹，则必须将其包含在路径中，并在配置文件中添加代码：GettingStarted/getting-started.py。

3. 运行以下命令：

   $ aws s3 --region aws region cp myapp.zip s3://ka-app-code-<username>

1. 打开 Managed Service for Apache Flink 控制台，网址为 https://console.aws.amazon.com/flink

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

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

   • 对于 应用程序名称 ，输入 MyApplication。

   • 对于描述，输入 My java test app。

   • 对于 运行时，请选择 Apache Flink。

   • 将版本保留为 Apache Flink 版本 1.15.2（建议的版本）。

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

5. 选择创建应用程序。

配置应用程序

1. 在MyApplication页面上，选择配置。

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

   • 对于Amazon S3 存储桶，请输入ka-app-code-<username>。

   • 在 Amazon S3 对象的路径中，输入myapp.zip。

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

4. 在属性下面，选择添加组。

5. 输入以下信息：

   组 ID	键	值
   consumer.config.0	input.stream.name	ExampleInputStream
   consumer.config.0	aws.region	us-west-2
   consumer.config.0	scan.stream.initpos	LATEST

   选择保存。

6. 在属性下面，再次选择添加组。

7. 输入以下信息：

   组 ID	键	值
   producer.config.0	output.stream.name	ExampleOutputStream
   producer.config.0	aws.region	us-west-2
   producer.config.0	shard.count	1

8. 在属性下面，再次选择添加组。对于 组 ID，输入 kinesis.analytics.flink.run.options。这个特殊的属性组告诉你的应用程序在哪里可以找到它的代码资源。有关更多信息，请参阅 指定您的代码文件。

9. 输入以下信息：

   组 ID	键	值
   kinesis.analytics.flink.run.options	python	getting-started.py
   kinesis.analytics.flink.run.options	jarfile	flink-sql-connector-kinesis-1.15.2.jar

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

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

12. 选择更新。

编辑 IAM policy 以添加 S3 存储桶权限

1. 通过 https://console.aws.amazon.com/iam/ 打开 IAM 控制台。

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

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

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

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/myapp.zip"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "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": "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"
           }
       ]
   }