示例：写入 Kinesis Data Firehose

1. 使用以下内容创建名为 stock.py 的文件：

    import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. 运行 stock.py 脚本：

   $ python stock.py

   在完成本教程的其余部分时，请将脚本保持运行状态。

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

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

2. 导航到 amazon-kinesis-data-analytics-java-examples/FirehoseSink 目录。

1. 如果还没有 Java 和 Maven，请安装它们。有关更多信息，请参阅入门指南 (DataStream API)教程中的先决条件。

2. 要将 Kinesis 连接器用于以下应用程序，您需要下载、构建并安装 Apache Maven。有关更多信息，请参阅将 Apache Flink Kinesis Streams 连接器与之前的 Apache Flink 版本一起使用。

3. 使用以下命令编译应用程序：

   mvn package -Dflink.version=1.15.3

   注意

   提供的源代码依赖于 Java 11 中的库。

上传应用程序代码

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

2. 在控制台中，选择 ka-app-code- <username>存储桶，然后选择上传。

3. 在选择文件步骤中，选择添加文件。导航到您在上一步中创建的 java-getting-started-1.0.jar 文件。

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

1. 打开 Managed Service for Apache Flink 控制台，网址为 https://console.aws.amazon.com/flink

2. 在 Managed Service for Apache Flink 控制面板上，选择创建分析应用程序。

3. 在Managed Service for Apache Flink - 创建应用程序页面上，提供应用程序详细信息，如下所示：

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

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

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

     注意

     Managed Service for Apache Flink 使用 Apache Flink 版本 1.15.2。

   • 将版本下拉列表保留为 Apache Flink 版本 1.15.2（建议的版本）。

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

5. 选择创建应用程序。

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

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

3. 在 摘要 页面上，选择 编辑策略。选择 JSON 选项卡。

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

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/java-getting-started-1.0.jar"
               ]
           },
           {
               "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"
           },
           {
               "Sid": "WriteDeliveryStream",
               "Effect": "Allow",
               "Action": "firehose:*",
               "Resource": "arn:aws:firehose:us-west-2:012345678901:deliverystream/ExampleDeliveryStream"
           }
       ]
   }

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

2. 在 配置应用程序 页面上，提供 代码位置：

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

   • 在 Amazon S3 对象的路径中，输入java-getting-started-1.0.jar。

3. 在 对应用程序的访问权限 下，对于 访问权限，选择 创建/更新 IAM 角色 kinesis-analytics-MyApplication-us-west-2。

4. 在 监控 下，确保 监控指标级别 设置为 应用程序。

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

6. 选择更新。

创建 IAM 角色

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

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

3. 在 选择受信任实体的类型 下，选择 Amazon 服务。在 选择将使用此角色的服务 下，选择 Kinesis。在选择您的使用案例下，选择 Kinesis Analytics。

   选择下一步: 权限。

4. 在 附加权限策略 页面上，选择 下一步: 审核。在创建角色后，您可以附加权限策略。

5. 在 创建角色 页面上，输入MF-stream-rw-role作为角色名称。选择 创建角色。

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

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

   注意

   对于本练习，Managed Service for Apache Flink代入此角色，以便同时从 Kinesis 数据流（源）读取数据和将输出写入另一个 Kinesis 数据流。因此，您附加在上一步（创建权限策略）中创建的策略。

   1. 在 摘要 页上，选择 权限 选项卡。

   2. 选择附加策略。

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

   4. 选择 A K ReadSourceStreamWriteSinkStream 策略，然后选择附加策略。

1. 将以下 JSON 代码保存到名为 create_request.json 的文件中。将示例角色 ARN 替换为您之前创建的角色的 ARN。将存储桶 ARN 后缀替换为在创建相关资源一节中选择的后缀 (ka-app-code-<username>)。将服务执行角色中的示例账户 ID (012345678901) 替换为您的账户 ID。

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_15",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "java-getting-started-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           }
         }
       }
   }
   
   

2. 使用上述请求执行 CreateApplication 操作来创建应用程序：

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

启动应用程序

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

   {
       "ApplicationName": "test",
       "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": "test"
   }
   

2. 使用下面的请求执行 StopApplication 操作来停止应用程序：

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

1. 打开 Managed Service for Apache Flink 控制台，网址为 https://console.aws.amazon.com/flink

2. 在 “适用于 Apache Flink 的托管服务” 面板中，选择。MyApplication

3. 选择配置。

4. 在 Snapshots (快照) 部分中，选择 Disable (禁用)，然后选择 Update (更新)。

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

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

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

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

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

2. 在 Kinesis Data Firehose 面板中，选择。ExampleDeliveryStream

3. 在ExampleDeliveryStream页面中，选择 “删除 Kinesis Data Fireh ose 直播”，然后确认删除。

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

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

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

4. 如果您为 Kinesis Data Firehose 流目标创建了 Amazon S3 存储桶，则也会删除该存储桶。

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

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

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

4. 选择 kinesis-analytics-service--us-MyApplication west-2 策略。

5. 选择 策略操作，然后选择 删除。

6. 如果您为 Kinesis Data Firehose 流创建了新策略，则也会删除该策略。

7. 在导航栏中，选择 角色。

8. 选择 k inesis-analytics-us-west-2 角色MyApplication。

9. 选择 删除角色，然后确认删除。

10. 如果您为 Kinesis Data Firehose 流创建了新角色，则也会删除该角色。

1. 打开 CloudWatch 控制台，网址为 https://console.aws.amazon.com/cloudwatch/。

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

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

4. 选择 删除日志组，然后确认删除。