创建并运行适用于 Apache Flink 的托管服务应用程序

创建数据流 (Amazon CLI)

1. 要创建第一个流（ExampleInputStream），请使用以下 Amazon Kinesis create-stream Amazon 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. 导航到 amazon-managed-service-for-apache-flink-examples/tree/main/java/GettingStarted 目录。

观察记录

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

2. 验证区域是否与运行本教程的区域相同，默认为 us-east-1 美国东部（弗吉尼亚州北部）。如果区域不匹配，请更改区域。

3. 选择数据流。

4. 选择您要观察的流，即 ExampleInputStream 或 ExampleOutputStream.

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

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

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

上传应用程序代码 JAR 文件

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

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

3. 选择上传。

4. 选择添加文件。

5. 导航到上一步中生成的 JAR 文件：target/amazon-msf-java-stream-app-1.0.jar。

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.20。

5. 配置应用程序

   • 应用程序名称：输入 MyApplication。

   • 描述：输入 My java test app。

   • 访问应用程序资源：选择使用所需策略创建/更新 IAM 角色 kinesis-analytics-MyApplication-us-east-1。

6. 配置用于应用程序设置的模板

   • 模板：选择开发。

7. 选择页面底部的创建流应用程序。

编辑策略

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 对象的路径中，输入 amazon-msf-java-stream-app-1.0.jar。

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

4. 在运行时属性部分中，添加以下属性。

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

   组 ID	键	值
   InputStream0	stream.name	ExampleInputStream
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1

6. 请勿修改任何其他部分。

7. 选择保存更改。

运行应用程序

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 向输入流发送数据。