步骤 3：创建并运行 Kinesis Data Analytics 应用程序

创建数据流 (Amazon CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesiscreate-streamAmazon CLI 命令。

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

2. 要创建应用程序用来写入输出的第二个流，请运行同一命令（将流名称更改为 ExampleOutputStream）。

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

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'))
   

2. 在本教程的后面部分，您运行 stock.py 脚本，以将数据发送到应用程序。

   $ python stock.py

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

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

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

编译应用程序代码

1. 要使用您的应用程序代码，您将其编译和打包成 JAR 文件。您可以通过两种方式之一编译和打包您的代码：

   • 使用命令行 Maven 工具。在包含 pom.xml 文件的目录中通过运行以下命令创建您的 JAR 文件：

     mvn package

     注意

     对于 Kinesis Data Analytics Runtime 版本 1.0.1，不需要-dflink.Version 参数；只有版本 1.1.0 及更高版本才需要该参数。有关更多信息，请参阅指定应用程序的 Apache Flink 版本：

   • 设置开发环境。有关详细信息，请参阅您的开发环境文档。

   您可以作为 JAR 文件上传您的包，也可以将包压缩为 ZIP 文件并上传。如果您使用 Amazon CLI 创建应用程序，您可以指定您的代码内容类型（JAR 或 ZIP）。

2. 如果编译时出错，请验证 JAVA_HOME 环境变量设置正确。

上传应用程序代码

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

2. 选择 Create bucket (创建存储桶)。

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

4. 在配置选项步骤中，让设置保持原样，然后选择下一步。

5. 在设置权限步骤中，让设置保持原样，然后选择下一步。

6. 选择创建桶。

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

8. 在选择文件步骤中，选择添加文件。导航到您在上一步中创建的 aws-kinesis-analytics-java-apps-1.0.jar 文件。选择下一步。

9. 在设置权限步骤中，让设置保持原样。选择下一步。

10. 在设置属性步骤中，让设置保持原样。请选择 Upload（上传）。

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

2. 在 Amazon Kinesis Data Analytics 控制面板上，选择创建分析应用程序。

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

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

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

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

     注意

     Kinesis Data Analytics 使用 Apache Flink 版本 1.8.2 或 1.6.2。

   • 将版本下拉列表更改为 Apache Flink 1.6。

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

5. 选择创建应用程序。

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: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": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

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

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

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

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

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

4. 输入以下应用程序属性和值：

   组 ID	键	值
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

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

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

7. 选择更新。

1. 在MyApplication页面上，选择 “运行”。确认该操作。

2. 当应用程序正在运行时，请刷新页面。控制台将显示 Application graph (应用程序图表)。

创建 IAM 角色

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

2. 在导航窗格中，选择 Roles (角色) 和 Create Role (创建角色)。

3. 在 “选择可信身份类型” 下，选择 “Amazon服务”。在 Choose the service that will use this role (选择将使用此角色的服务) 下，选择 Kinesis。在选择您的使用案例下，选择 Kinesis Analytics。

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

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

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

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

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

   注意

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

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

   2. 选择附加策略。

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

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

1. 将以下 JSON 代码保存到名为 create_request.json 的文件中。将示例角色 ARN 替换为您之前创建的角色的 ARN。将存储桶 ARN 后缀 (username) 替换为在前一部分中选择的后缀。将服务执行角色中的示例账户 ID (012345678901) 替换为您的账户 ID。

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_6",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/KA-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "java-getting-started-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

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. 将以下 JSON 代码保存到名为 update_properties_request.json 的文件中。

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

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

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