创建并运行面向 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-east-1 

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

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-east-1 

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

   git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git

2. 导航到 ./python/GettingStarted 目录。

要运行应用程序，请从命令行运行

1. 确保在其中安装 Python Flink 库的 Conda 或 VirtuaLenv 等独立 Python 环境当前处于激活状态。

2. 确保至少运行一次 mvn package。

3. 设置 IS_LOCAL = true 环境变量：

   $ export IS_LOCAL=true

4. 将该应用程序作为常规 Python 脚本运行。

   $python main.py

从 IDE 中运行应用程序

1. 使用以下配置将 IDE 配置为运行 main.py 脚本：

   1. 使用独立的 Python 环境，例如在其中安装 PyfLink 库的 Conda 或 VirtualenV。

   2. 使用 Amazon 凭证访问输入和输出 Kinesis 数据流。

   3. 设置 IS_LOCAL = true。

2. 设置运行配置的确切过程取决于您的 IDE，并且会有所不同。

3. 设置 IDE 后，运行 Python 脚本，并在应用程序运行时使用 IDE 提供的工具。

上传应用程序代码 JAR 文件

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

2. 选择您之前为应用程序代码创建的存储桶。

3. 选择上传。

4. 选择添加文件。

5. 导航到上一步中生成的 .zip 文件：target/managed-flink-pyflink-getting-started-1.0.0.zip。

6. 在不更改任何其他设置的情况下选择上传。

1. 登录 Amazon Web Services 管理控制台 并打开 Amazon MSF 控制台，网址为 https://console.aws.amazon.com/flink。

2. 确认选择正确的区域：美国东部（弗吉尼亚北部）us-east-1。

3. 打开右侧的菜单，选择 Apache Flink 应用程序，然后选择创建流应用程序。或者，从初始页面的入门部分中选择创建流应用程序。

4. 在创建流应用程序页面上：

   • 在选择设置流处理应用程序的方法中，选择从头开始创建。

   • 对于 Apache Flink 配置，应用程序 Flink 版本，请选择 Apache Flink 1.19。

   • 对于应用程序配置：

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

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

     • 在对应用程序资源的访问权限中，选择创建/更新具有所需策略的 IAM 角色 Kinesis-analytics-myApplication-us-east-1。

   • 对于应用程序模板设置：

     • 对于模板，选择开发。

   • 选择创建流应用程序。

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

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

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

3. 选择编辑，然后选择 JSON 选项卡。

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

   JSON

   {
       "Version":"2012-10-17",		 	 	 
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::my-bucket/kinesis-analytics-placeholder-s3-object"
               ]
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleOutputStream"
           }
       ]
   }
   

5. 选择下一步，然后选择保存更改。

配置应用程序

1. 在我的应用程序 页面上，选择配置。

2. 在应用程序代码位置部分中：

   • 对于 Amazon S3 存储桶，请选择之前为应用程序代码创建的存储桶。选择浏览并选择正确的存储桶，然后选择选择。请勿在存储桶名称上选择。

   • 在 Amazon S3 对象的路径中，输入 managed-flink-pyflink-getting-started-1.0.0.zip。

3. 对于访问权限，请选择创建/更新具有必要策略的 IAM 角色 kinesis-analytics-MyApplication-us-east-1。

4. 移至运行时属性，并保留所有其他设置的默认值。

5. 选择添加新项目并添加以下每个参数：

   组 ID	键	值
   InputStream0	stream.name	ExampleInputStream
   InputStream0	flink.stream.initpos	LATEST
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1
   kinesis.analytics.flink.run.options	python	main.py
   kinesis.analytics.flink.run.options	jarfile	lib/pyflink-dependencies.jar

6. 请勿修改任何其他部分，然后选择保存更改。

运行应用程序

1. 在 Amazon Managed Service for Apache Flink 的控制台上，选择我的应用程序，然后选择运行。

2. 在下一页的“应用程序还原配置”页面上，选择使用最新快照运行，然后选择运行。

   应用程序详细信息中的状态会从 Ready 转换到 Starting，然后在应用程序启动时转换到 Running。

打开控制面板

1. 选择打开 Apache Flink 控制面板。控制面板将在新页面上打开。

2. 在正在运行的作业列表中，选择您可以看到的单个作业。

   注意

   如果您错误设置运行时属性或编辑 IAM 策略，则应用程序状态可能会变为 Running，但是 Flink 控制面板显示任务正在持续重新启动。如果应用程序配置错误或缺乏访问外部资源的权限，则通常会出现这种故障场景。

   发生这种情况时，请检查 Flink 控制面板中的异常选项卡以查看问题的原因。

查看指标

1. 在刷新按钮旁边，从下拉列表中选择 10 秒。

2. 当应用程序运行且运行状况良好时，您可以看到正常运行时间指标不断增加。

3. 完全重新启动指标应为零。如果该指标增加，则配置可能存在问题。要调查问题，请审查 Flink 控制面板上的异常选项卡。

4. 在运行状况良好的应用程序中，失败的检查点数指标应为零。

   注意

   此控制面板显示一组固定的指标，且粒度为 5 分钟。您可以在 CloudWatch 控制面板中创建包含任何指标的自定义应用程序控制面板。

查看输出

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

2. 确认该区域与您运行本教程时使用的区域相同。默认情况下，区域为 US-East-1US 东部（弗吉尼亚北部）。如有必要，可以更改区域。

3. 选择数据流。

4. 选择您要观察的流。在本教程中，请使用 ExampleOutputStream。

5. 选择数据查看器选项卡。

6. 选择任意分片，保持最新作为起始位置，然后选择获取记录。您可能会看到“未找到该请求的记录”错误。如果看到此错误，请选择重试获取记录。发布到流显示的最新记录。

7. 在“数据”列中选择值以检查 JSON 格式的记录内容。

停止应用程序

1. 从操作下拉列表中，选择停止。

2. 应用程序详细信息中的状态会从 Running 转换到 Stopping，然后在应用程序完全停止时转换到 Ready。

   注意

   请勿忘记还要停止从 Python 脚本或 Kinesis Data Generator 向输入流发送数据。