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

下载 Java 应用程序代码

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

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

2. 导航到 ./java/GettingStartedTable 目录。

浏览 S3 存储桶并观察应用程序写入的文件

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

2. 选择您之前创建的存储桶。

3. 导航到 output 路径，然后导航到与 UTC 时区当前时间相对应的日期和小时文件夹。

4. 定期刷新以观察每 5 秒钟出现的新文件。

5. 选择并下载一个文件以观察其内容。

   注意

   默认情况下，这些文件没有扩展名。该内容的格式设置为 JSON。您可以使用任何文本编辑器打开文件来检查内容。

上传应用程序代码

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

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

3. 选择上传字段。

4. 选择添加文件。

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

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

   警告

   确保在 <repo-dir>/java/GettingStarted/target/amazon/msf-java-table-app-1.0.jar 中选择正确的 JAR 文件。

   目标目录还包含您无需上传的其他 JAR 文件。

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 Java Table API test app。

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

   • 在应用程序设置模板中，完成以下操作：

     • 对于模板，请选择开发。

5. 选择创建流应用程序。

编辑 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，将 <bucket-name> 替换为您创建的 S3 存储桶的名称。

   JSON

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

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

配置应用程序

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

2. 在应用程序代码位置部分，选择配置。

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

   • 在 Amazon S3 对象的路径中，输入 amazon-msf-java-table-app-1.0.jar。

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

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

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

   组 ID	键	值
   bucket	name	your-bucket-name
   bucket	path	output

6. 请勿修改任何其他设置。

7. 选择保存更改。

运行应用程序

1. 返回 Amazon Managed Service for Apache Flink 中的控制台页面，然后选择我的应用程序。

2. 选择运行以启动应用程序。

3. 在应用程序还原配置中，选择使用最新快照运行。

4. 选择运行。

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

打开控制面板并查看作业

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

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

   注意

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

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

查看指标

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

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

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

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

   注意

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

停止应用程序

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

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

   注意

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