创建并运行适用于 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 上打开 Kinesis 控制台。https://console.aws.amazon.com

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. 选择 Add files。

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

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

创建应用程序

1. 登录并通过 /f Amazon Web Services Management Console link 打开亚马逊 MSF 控制台。 https://console.aws.amazon.com

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

3. 打开右侧的菜单，选择 Apache Flink 应用程序，然后选择 “创建流媒体应用程序”。或者，在初始页面的 “入门” 容器中选择 “创建流媒体应用程序”。

4. 在 “创建流媒体应用程序” 页面上：

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

   • Apache Flink 配置，Application Flink 版本：选择 Ap ache 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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (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. 在MyApplication页面上，选择配置。

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. 在适用于 Apache Flink 的亚马逊托管服务的控制台上，选择 “我的应用程序”，然后选择 “运行”。

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

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

打开 控制面板

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

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

   注意

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

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

查看指标

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

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

3. 完全重启指标应为零。如果它增加，则配置可能会出现问题。要调查问题，请查看 Flink 控制面板上的 “异常” 选项卡。

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

   注意

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

查看输出

1. 在 /kinesis 上打开 Kinesis 控制台。https://console.aws.amazon.com

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

3. 选择 “数据流”。

4. 选择要观看的直播。在本教程中，请使用 ExampleOutputStream。

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

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

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

停止应用程序

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

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

   注意

   别忘了停止从 Python 脚本或 Kinesis 数据生成器向输入流发送数据。