示例：在 Scala 中将流媒体数据发送到 Amazon S3

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

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

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

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

编译应用程序代码

使用 SBT 编译工具为应用程序构建 Scala 代码。要安装 SBT，请参阅使用 cs 设置安装 sbt。您还需要安装 Java 开发工具包 (JDK)。请参阅完成练习的先决条件。

1. 要使用您的应用程序代码，您将其编译和打包成 JAR 文件。你可以用 SBT 编译和打包你的代码：

   sbt assembly

2. 如果应用程序成功编译，则创建以下文件：

   target/scala-3.2.0/s3-sink-scala-1.0.jar

上传 Apache Flink 直播 Scala 代码

在本节中，您将创建一个 Simple Storage S3 存储桶并上传您的应用程序代码。

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

2. 选择创建存储桶

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

4. 在配置选项中，保持设置不变，然后选择下一步。

5. 在设置权限中，保持设置不变，然后选择下一步。

6. 选择创建桶。

7. 选择ka-app-code-<username>存储桶然后选择 Upload（上载）。

8. 在选择文件步骤中，选择添加文件。导航到您在上一步中创建的 s3-sink-scala-1.0.jar 文件。

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

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

2. 在 Kinesis Data Analytics 仪表板上，选择创建分析应用程序。

3. 在 Kinesis Analytics – 创建应用程序页面上，提供应用程序详细信息，如下所示：

   • 对于 Application name (应用程序名称)，输入 MyApplication。

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

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

   • 将该版本保留为 Apache Flink 版本 1.15.2（推荐版本）。

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

5. 选择创建应用程序。

配置应用程序

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

2. 在 Configure application (配置应用程序) 页面上，提供 Code location (代码位置)：

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

   • 在 Amazon S3 对象的路径中，输入s3-sink-scala-1.0.jar。

3. 在 Access to application resources (对应用程序的访问权限) 下，对于 Access permissions (访问权限)，选择 Create / update IAM role (创建/更新 IAM 角色) kinesis-analytics-MyApplication-us-west-2。

4. 在 “属性” 下，选择 “添加群组”。

5. 输入以下信息:

   组 ID	键	值
   ConsumerConfigProperties	input.stream.name	ExampleInputStream
   ConsumerConfigProperties	aws.region	us-west-2
   ConsumerConfigProperties	flink.stream.initpos	LATEST

   选择保存。

6. 在 “属性” 下，选择 “添加群组”。

7. 输入以下信息:

   组 ID	键	值
   ProducerConfigProperties	s3.sink.path	s3a://ka-app-code-<user-name>/data

8. 在 Monitoring (监控) 下，确保 Monitoring metrics level (监控指标级别) 设置为 Application (应用程序)。

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

10. 选择更新。

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

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

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

3. 在 Summary (摘要) 页面上，选择 Edit policy (编辑策略)。请选择 JSON 选项卡。

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

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:Abort*", 
                   "s3:DeleteObject*", 
                   "s3:GetObject*", 
                   "s3:GetBucket*", 
                   "s3:List*", 
                   "s3:ListBucket", 
                   "s3:PutObject"
               ],
               "Resource": [ 
                   "arn:aws:s3:::ka-app-code-<username>", 
                   "arn:aws:s3:::ka-app-code-<username>/*" 
               ]
           },
           {
               "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"
           }
       ]
   }

创建 IAM 角色

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

2. 在导航窗格中，选择角色，然后选择创建角色。

3. 在 “选择可信身份类型” 下，选择 “Amazon服务”

4. 在 Choose the service that will use this role (选择将使用此角色的服务) 下，选择 Kinesis。

5. 在选择您的使用案例下，选择 Kinesis Analytics。

6. 选择Next: Permissions（下一步: 权限）。

7. 在 Attach permissions policies 页面上，选择 Next: Review。在创建角色后，您可以附加权限策略。

8. 在创建角色页面上KA-stream-rw-role，输入角色名称。选择 Create role（创建角色）。

   现在，您已经创建了一个名为的新 IAM 角色KA-stream-rw-role。接下来，更新角色的信任和权限策略

9. 将权限策略附加到角色。

   注意

   在本练习中，Kinesis Data Analytics 扮演这个角色，既可以从 Kinesis 数据流（源）读取数据，也可以将输出写入另一个 Kinesis 数据流。因此，您可以附加您在上一步中创建的策略，即创建权限策略。

   1. 在 Summary (摘要) 页上，选择 Permissions (权限) 选项卡。

   2. 选择附加策略。

   3. 在搜索框中，输入 KAReadSourceStreamWriteSinkStream（您在上一部分中创建的策略）。

   4. 选择KAReadSourceStreamWriteSinkStream策略，然后选择附加策略。

启动应用程序

1. 将以下 JSON 代码保存到名为 start_request.json 的文件中。

   {{
       "ApplicationName": "s3_sink",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }

2. 使用上述请求执行 StartApplication 操作来启动应用程序：

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

停止应用程序

1. 将以下 JSON 代码保存到名为 stop_request.json 的文件中。

   {
      "ApplicationName": "s3_sink"
   }

2. 使用上述请求执行StopApplication操作以停止应用程序：

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

更新应用程序的环境属性

1. 将以下 JSON 代码保存到名为 update_properties_request.json 的文件中。

   {"ApplicationName": "s3_sink",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2",
                       "stream.name" : "ExampleInputStream",
                       "flink.stream.initpos" : "LATEST"
                  }
               },
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "s3.sink.path" : "s3a://ka-app-code-<username>/data"
                  }
               }
            ]
         }
      }
   }

2. 使用前面的请求执行 UpdateApplication 操作以更新环境属性：

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. 在 Kinesis Data Analytics 面板中，选择MyApplication。

3. 在应用程序的页面中，选择 Delete (删除)，然后确认删除。

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

2. 在 Kinesis Data Streams 面板中，选择ExampleInputStream。

3. 在ExampleInputStream页面中，选择 “删除 Kinesis Stream”，然后确认删除。

4. 在 Kinesis stream s 页面中 ExampleOutputStream，选择，选择操作，选择删除，然后确认删除。

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

2. 选择 ka-app-code- 存储桶。 <username>

3. 选择 Delete (删除)，然后输入存储桶名称以确认删除。

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

2. 在导航栏中，选择策略。

3. 在筛选条件控件中，输入 kinesis。

4. 选择 kinesis-analytics-service-MyApplication- <your-region>策略。

5. 选择 Policy Actions (策略操作)，然后选择 Delete (删除)。

6. 在导航栏中，选择 Roles（角色）。

7. 选择 k inesis-analyticsMyApplication- <your-region>角色。

8. 选择 Delete role (删除角色)，然后确认删除。

1. 通过 https://console.aws.amazon.com/cloudwatch/ 打开 CloudWatch 主机。

2. 在导航栏中，选择 Logs (日志)。

3. 选择 /aws/kinesis-analytics/MyApplication 日志组。

4. 选择 Delete Log Group (删除日志组)，然后确认删除。