适用于 Apache Flink 的托管服务的早期版本信息

1. 确保已安装 Apache Maven，并且 JAVA_HOME 环境变量指向 JDK 而不是 JRE。您可以使用以下命令测试 Apache Maven 安装：

   mvn -version

2. 下载 Apache Flink 版本 1.8.2 源代码：

   wget https://archive.apache.org/dist/flink/flink-1.8.2/flink-1.8.2-src.tgz

3. 解压缩 Apache Flink 源代码：

   tar -xvf flink-1.8.2-src.tgz

4. 转到 Apache Flink 源代码目录：

   cd flink-1.8.2

5. 编译并安装 Apache Flink：

   mvn clean install -Pinclude-kinesis -DskipTests

   注意

   如果您在微软 Windows 上编译 Flink，则需要添加-Drat.skip=true参数。

要注册 Amazon Web Services 账户

1. 打开https://portal.aws.amazon.com/billing/注册。

2. 按照屏幕上的说明操作。

   在注册时，将接到电话或收到短信，要求使用电话键盘输入一个验证码。

   当您注册时 Amazon Web Services 账户，就会创建Amazon Web Services 账户根用户一个。根用户有权访问该账户中的所有 Amazon Web Services 服务 和资源。作为最佳安全实践，请为用户分配管理访问权限，并且只使用根用户来执行需要根用户访问权限的任务。

要设置 Amazon CLI

1. 下载并配置 Amazon CLI。有关说明，请参阅《Amazon Command Line Interface 用户指南》中的以下主题：

   • 安装 Amazon Command Line Interface

   • 配置 Amazon CLI

2. 在文件中为管理员用户添加已命名的配置 Amazon CLI config文件。在执行 Amazon CLI 命令时，您将使用此配置文件。有关命名配置文件的更多信息，请参阅 Amazon Command Line Interface 用户指南中的命名配置文件。

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   有关可用 Amazon 区域的列表，请参阅中的区域和终端节点Amazon Web Services 一般参考。

   注意

   本教程中的示例代码和命令使用美国西部（俄勒冈州）区域。要使用不同的区域，请将本教程的代码和命令中的区域更改为要使用的区域。

3. 在命令提示符处输入以下帮助命令来验证设置：

   aws help

创建数据流 (Amazon CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesis 命令create-stream Amazon 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', 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/GettingStarted 目录。

编译应用程序代码

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

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

     mvn package -Dflink.version=1.13.2

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

     注意

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

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

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

上传应用程序代码

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

2. 选择 创建存储桶 。

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. 您无需更改该对象的任何设置，因此，请选择 上传。

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

   • 将版本下拉列表保留为 Apache Flink 版本 1.13。

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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (012345678901) 替换为您的账户 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-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. 在 配置应用程序 页面上，提供 代码位置：

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

   • 在 Amazon S3 对象的路径中，输入aws-kinesis-analytics-java-apps-1.0.jar。

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

4. 输入以下信息：

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

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

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

7. 选择更新。

创建 IAM 角色

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

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

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

   选择下一步: 权限。

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

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

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

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

   注意

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

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

   2. 选择附加策略。

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

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

1. 将以下 JSON 代码保存到名为 create_request.json 的文件中。将示例角色 ARN 替换为您之前创建的角色的 ARN。将存储桶 ARN 后缀 (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": "aws-kinesis-analytics-java-apps-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

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

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

要注册 Amazon Web Services 账户

1. 打开https://portal.aws.amazon.com/billing/注册。

2. 按照屏幕上的说明操作。

   在注册时，将接到电话或收到短信，要求使用电话键盘输入一个验证码。

   当您注册时 Amazon Web Services 账户，就会创建Amazon Web Services 账户根用户一个。根用户有权访问该账户中的所有 Amazon Web Services 服务 和资源。作为最佳安全实践，请为用户分配管理访问权限，并且只使用根用户来执行需要根用户访问权限的任务。

要设置 Amazon CLI

1. 下载并配置 Amazon CLI。有关说明，请参阅《Amazon Command Line Interface 用户指南》中的以下主题：

   • 安装 Amazon Command Line Interface

   • 配置 Amazon CLI

2. 在文件中为管理员用户添加已命名的配置 Amazon CLI config文件。在执行 Amazon CLI 命令时，您将使用此配置文件。有关命名配置文件的更多信息，请参阅 Amazon Command Line Interface 用户指南中的命名配置文件。

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   有关可用 Amazon 区域的列表，请参阅中的区域和终端节点Amazon Web Services 一般参考。

   注意

   本教程中的示例代码和命令使用美国西部（俄勒冈州）区域。要使用不同的区域，请将本教程的代码和命令中的区域更改为要使用的区域。

3. 在命令提示符处输入以下帮助命令来验证设置：

   aws help

创建数据流 (Amazon CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesis 命令create-stream Amazon 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-examples.git

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

编译应用程序代码

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

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

     mvn package -Dflink.version=1.11.3

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

     注意

     提供的源代码依赖于 Java 11 中的库。确保项目的 Java 版本为 11。

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

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

上传应用程序代码

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

2. 选择 创建存储桶 。

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. 您无需更改该对象的任何设置，因此，请选择 上传。

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

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

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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (012345678901) 替换为您的账户 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-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. 在 配置应用程序 页面上，提供 代码位置：

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

   • 在 Amazon S3 对象的路径中，输入aws-kinesis-analytics-java-apps-1.0.jar。

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

4. 在 Properties (属性) 下，对于 Group ID (组 ID)，输入 ProducerConfigProperties。

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

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

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

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

8. 选择更新。

创建 IAM 角色

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

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

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

   选择下一步: 权限。

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

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

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

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

   注意

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

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

   2. 选择附加策略。

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

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

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

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_11",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-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

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

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

要注册 Amazon Web Services 账户

1. 打开https://portal.aws.amazon.com/billing/注册。

2. 按照屏幕上的说明操作。

   在注册时，将接到电话或收到短信，要求使用电话键盘输入一个验证码。

   当您注册时 Amazon Web Services 账户，就会创建Amazon Web Services 账户根用户一个。根用户有权访问该账户中的所有 Amazon Web Services 服务 和资源。作为最佳安全实践，请为用户分配管理访问权限，并且只使用根用户来执行需要根用户访问权限的任务。

要设置 Amazon CLI

1. 下载并配置 Amazon CLI。有关说明，请参阅《Amazon Command Line Interface 用户指南》中的以下主题：

   • 安装 Amazon Command Line Interface

   • 配置 Amazon CLI

2. 在文件中为管理员用户添加已命名的配置 Amazon CLI config文件。在执行 Amazon CLI 命令时，您将使用此配置文件。有关命名配置文件的更多信息，请参阅 Amazon Command Line Interface 用户指南中的命名配置文件。

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   有关可用区域的列表，请参阅 Amazon Web Services 一般参考中的区域和终端节点。

   注意

   本教程中的示例代码和命令使用美国西部（俄勒冈州）区域。要使用其他 Amazon 区域，请将本教程的代码和命令中的区域更改为要使用的区域。

3. 在命令提示符处输入以下帮助命令来验证设置：

   aws help

创建数据流 (Amazon CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesis 命令create-stream Amazon 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-examples.git

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

编译应用程序代码

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

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

     mvn package -Dflink.version=1.8.2

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

     注意

     提供的源代码依赖于 Java 1.8 中的库。确保项目的 Java 版本为 1.8。

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

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

上传应用程序代码

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

2. 选择 创建存储桶 。

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. 您无需更改该对象的任何设置，因此，请选择 上传。

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

   • 将版本下拉列表保留为 Apache Flink 1.8 (Recommended Version) (Apache Flink 1.8 (建议的版本))。

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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (012345678901) 替换为您的账户 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-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. 在 配置应用程序 页面上，提供 代码位置：

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

   • 在 Amazon S3 对象的路径中，输入aws-kinesis-analytics-java-apps-1.0.jar。

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

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

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

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

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

7. 选择更新。

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

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

创建 IAM 角色

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

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

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

   选择下一步: 权限。

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

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

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

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

   注意

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

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

   2. 选择附加策略。

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

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

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

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_8",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-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

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

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

3. 选择 配置。

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

要注册 Amazon Web Services 账户

1. 打开https://portal.aws.amazon.com/billing/注册。

2. 按照屏幕上的说明操作。

   在注册时，将接到电话或收到短信，要求使用电话键盘输入一个验证码。

   当您注册时 Amazon Web Services 账户，就会创建Amazon Web Services 账户根用户一个。根用户有权访问该账户中的所有 Amazon Web Services 服务 和资源。作为最佳安全实践，请为用户分配管理访问权限，并且只使用根用户来执行需要根用户访问权限的任务。

要设置 Amazon CLI

1. 下载并配置 Amazon CLI。有关说明，请参阅《Amazon Command Line Interface 用户指南》中的以下主题：

   • 安装 Amazon Command Line Interface

   • 配置 Amazon CLI

2. 在文件中为管理员用户添加已命名的配置 Amazon CLI config文件。在执行 Amazon CLI 命令时，您将使用此配置文件。有关命名配置文件的更多信息，请参阅 Amazon Command Line Interface 用户指南中的命名配置文件。

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   有关可用 Amazon 区域的列表，请参阅中的区域和终端节点Amazon Web Services 一般参考。

   注意

   本教程中的示例代码和命令使用美国西部（俄勒冈州）区域。要使用不同的区域，请将本教程的代码和命令中的区域更改为要使用的区域。

3. 在命令提示符处输入以下帮助命令来验证设置：

   aws help

创建数据流 (Amazon CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesis 命令create-stream Amazon 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-examples.git

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

编译应用程序代码

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

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

     mvn package

     注意

     Managed Service for Apache Flink 运行时系统版本 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. 选择 创建存储桶 。

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. 在设置属性步骤中，让设置保持原样。选择上传。

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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.8.2 或 1.6.2。

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

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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (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. 在 配置应用程序 页面上，提供 代码位置：

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

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

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

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

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

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

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

7. 选择更新。

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

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

创建 IAM 角色

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

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

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

   选择下一步: 权限。

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

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

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

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

   注意

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

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

   2. 选择附加策略。

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

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

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/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"
           },
           "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

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

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

3. 选择 配置。

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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 客户端，请安装它。有关更多信息，请参阅安装 Git。

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

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

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

1. 如果还没有 Java 和 Maven，请安装它们。有关更多信息，请参阅教程：开始使用适用于 Apache Flink 的托管服务中的 DataStream API教程中的完成必需的先决条件。

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

   mvn package -Dflink.version=1.15.3

   注意

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

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

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

   • 对于运行时系统，请选择 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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (012345678901) 替换为您的账户 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "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": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "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. 在 配置应用程序 页面上，提供 代码位置：

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

   • 在 Amazon S3 对象的路径中，输入aws-kinesis-analytics-java-apps-1.0.jar。

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

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

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

6. 选择更新。

1. 在MyApplication页面上，选择 “运行”。保持不使用快照运行选项处于选中状态，然后确认操作。

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

2. 在 Apache Flink 的托管服务面板中，选择。MyApplication

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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 客户端，请安装它。有关更多信息，请参阅安装 Git。

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

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

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

1. 如果还没有 Java 和 Maven，请安装它们。有关更多信息，请参阅教程：开始使用适用于 Apache Flink 的托管服务中的 DataStream API教程中的完成必需的先决条件。

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

   mvn package -Dflink.version=1.15.3

   注意

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

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

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

   • 将版本下拉列表保留为 Apache Flink 1.15.2 (Recommended Version) (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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (012345678901) 替换为您的账户 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "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": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "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. 在 配置应用程序 页面上，提供 代码位置：

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

   • 在 Amazon S3 对象的路径中，输入aws-kinesis-analytics-java-apps-1.0.jar。

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

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

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

6. 选择更新。

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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 客户端，请安装它。有关更多信息，请参阅安装 Git。

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

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

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

1. 如果还没有 Java 和 Maven，请安装它们。有关更多信息，请参阅教程：开始使用适用于 Apache Flink 的托管服务中的 DataStream API教程中的完成必需的先决条件。

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

   mvn package -Dflink.version=1.15.3

1. 在 Amazon S3 控制台中，选择 ka-app-code-<username> 存储桶，导航到代码文件夹，然后选择上传。

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

   • 将版本下拉列表保留为 Apache Flink 1.15.2 (Recommended Version) (Apache Flink 1.15.2 (建议的版本))。

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

5. 选择创建应用程序。

   注意

   在使用控制台创建应用程序的 Managed Service for Apache Flink时，您可以选择为应用程序创建 IAM 角色和策略。您的应用程序使用此角色和策略访问其从属资源。这些 IAM 资源是使用您的应用程序名称和区域命名的，如下所示：

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

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

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

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

7. 选择创建应用程序。

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

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

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

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

   {
               "Sid": "S3",
               "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": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:%LOG_GROUP_PLACEHOLDER%:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:%LOG_GROUP_PLACEHOLDER%:log-stream:%LOG_STREAM_PLACEHOLDER%"
               ]
           }
           ,
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           
       ]
   }

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

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

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

   • 在 Amazon S3 对象的路径中，输入code/aws-kinesis-analytics-java-apps-1.0.jar。

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

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

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

6. 选择更新。

1. 在MyApplication页面上，选择 “运行”。保持不使用快照运行选项处于选中状态，然后确认操作。

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

1. 在S3StreamingSinkJob.java文件开头添加以下导入语句：

   import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
   import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;

2. 更新代码中的createS3SinkFromStaticConfig()方法，使其看起来与以下内容类似：

   private static StreamingFileSink<String> createS3SinkFromStaticConfig() {
   
       final StreamingFileSink<String> sink = StreamingFileSink
           .forRowFormat(new Path(s3SinkPath), new SimpleStringEncoder<String>("UTF-8"))
           .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd--HH"))
           .withRollingPolicy(DefaultRollingPolicy.create().build())
           .build();
       return sink;
   }
               

1. 增加应用程序的CheckpointInterval设置。以下 UpdateApplication操作输入将检查点间隔设置为 10 分钟：

   {
      "ApplicationConfigurationUpdate": {  
         "FlinkApplicationConfigurationUpdate": { 
            "CheckpointConfigurationUpdate": {
               "ConfigurationTypeUpdate" : "CUSTOM", 
               "CheckpointIntervalUpdate": 600000
            }
         }
      },
      "ApplicationName": "MyApplication",
      "CurrentApplicationVersionId": 5
   }

   要使用上述代码，请指定当前应用程序版本。您可以使用ListApplications操作检索应用程序版本。

2. 在S3StreamingSinkJob.java文件开头添加以下导入语句：

   import java.util.concurrent.TimeUnit; 

3. 更新S3StreamingSinkJob.java文件中的createS3SinkFromStaticConfig方法，使其看起来与以下内容类似：

   private static StreamingFileSink<String> createS3SinkFromStaticConfig() {
   
           final StreamingFileSink<String> sink = StreamingFileSink
                   .forRowFormat(new Path(s3SinkPath), new SimpleStringEncoder<String>("UTF-8"))
                   .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd--HH"))
                   .withRollingPolicy(
                           DefaultRollingPolicy.create()
                               .withRolloverInterval(TimeUnit.MINUTES.toMillis(8))
                               .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                               .withMaxPartSize(1024 * 1024 * 1024)
                               .build())
                   .build();
           return sink;
       }

   前面的代码示例将写入 Amazon S3 存储桶的频率设置为 8 分钟。

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

3. 应用程序代码位于 amazon-kinesis-data-analytics-java-examples/KafkaConnectors/KafkaGettingStartedJob.java 文件中。您可以检查代码以熟悉Managed Service for Apache Flink的应用程序代码的结构。

4. 使用命令行 Maven 工具或首选的开发环境以创建 JAR 文件。要使用命令行 Maven 工具编译 JAR 文件，请输入以下内容：

   mvn package -Dflink.version=1.15.3

   如果构建成功，则会创建以下文件：

   target/KafkaGettingStartedJob-1.0.jar

   注意

   提供的源代码依赖于 Java 11 中的库。如果您使用的是开发环境，

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

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台。 https://console.aws.amazon.com

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

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

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

   • 对于运行时系统，请选择 Apache Flink 版本 1.15.2。

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

5. 选择创建应用程序。

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

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

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

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

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

   注意

   当您使用控制台指定应用程序资源（例如 CloudWatch 日志或 Amazon VPC）时，控制台会修改您的应用程序执行角色以授予访问这些资源的权限。

4. 在 Properties (属性) 下面，选择 Add Group (添加组)。输入以下属性：

   组 ID	键	值
   KafkaSource	topic	Amazon KafkaTutorialTopic
   KafkaSource	bootstrap.servers	The bootstrap server list you saved previously
   KafkaSource	security.protocol	SSL
   KafkaSource	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   KafkaSource	ssl.truststore.password	changeit

   注意

   默认证书的 ssl.truststore.password 为“changeit”；如果使用默认证书，则不需要更改该值。

   再次选择 Add Group (添加组)。输入以下属性：

   组 ID	键	值
   KafkaSink	topic	Amazon KafkaTutorialTopicDestination
   KafkaSink	bootstrap.servers	The bootstrap server list you saved previously
   KafkaSink	security.protocol	SSL
   KafkaSink	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   KafkaSink	ssl.truststore.password	changeit
   KafkaSink	transaction.timeout.ms	1000

   应用程序代码读取上述应用程序属性，以配置用于与 VPC 和 Amazon MSK 集群交互的源和接收器。有关使用属性的更多信息，请参阅使用运行时属性。

5. 在 Snapshots (快照) 下面，选择 Disable (禁用)。这样，就可以轻松更新应用程序，而无需加载无效的应用程序状态数据。

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

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

8. 在 Virtual Private Cloud (VPC) 部分中，选择要与应用程序关联的 VPC。选择与您的 VPC 关联的子网和安全组，您希望应用程序使用它们访问 VPC 资源。

9. 选择更新。

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 客户端，请安装它。有关更多信息，请参阅安装 Git。

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

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

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

1. 如果还没有 Java 和 Maven，请安装它们。有关更多信息，请参阅教程：开始使用适用于 Apache Flink 的托管服务中的 DataStream API教程中的完成必需的先决条件。

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

   mvn package -Dflink.version=1.15.3

   注意

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

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

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

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

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

   • 对于运行时系统，请选择 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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (012345678901) 替换为您的账户 ID。

   注意

   这些权限使应用程序能够访问 EFO 使用者。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "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": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "AllStreams",
               "Effect": "Allow",
               "Action": [
                   "kinesis:ListShards",
                   "kinesis:ListStreamConsumers",
                   "kinesis:DescribeStreamSummary"
               ],
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/*"
           },
           {
               "Sid": "Stream",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStream",
                   "kinesis:RegisterStreamConsumer",
                   "kinesis:DeregisterStreamConsumer"
               ],
               "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"
           },
           {
               "Sid": "Consumer",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:SubscribeToShard"
               ],
               "Resource": [
                   "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app",
                   "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app:*"
               ]
           }
       ]
   }

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

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

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

   • 在 Amazon S3 对象的路径中，输入aws-kinesis-analytics-java-apps-1.0.jar。

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

4. 在属性下，选择创建组。

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

   组 ID	键	值
   ConsumerConfigProperties	flink.stream.recordpublisher	EFO
   ConsumerConfigProperties	flink.stream.efo.consumername	basic-efo-flink-app
   ConsumerConfigProperties	INPUT_STREAM	ExampleInputStream
   ConsumerConfigProperties	flink.inputstream.initpos	LATEST
   ConsumerConfigProperties	AWS_REGION	us-west-2

6. 在属性下，选择创建组。

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

   组 ID	键	值
   ProducerConfigProperties	OUTPUT_STREAM	ExampleOutputStream
   ProducerConfigProperties	AWS_REGION	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

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

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

10. 选择更新。

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

2. 在 Apache Flink 的托管服务面板中，选择。MyApplication

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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，请安装它们。有关更多信息，请参阅教程：开始使用适用于 Apache Flink 的托管服务中的 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> 存储桶，然后选择 Uploa d。

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

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

1. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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. 将以下策略示例中突出显示的部分添加到策略中。将示例账户 IDs (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. 使用 https://console.aws.amazon.com/iam/ 打开 IAM 控制台。

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

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

   选择下一步: 权限。

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

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

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

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

   注意

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

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

   2. 选择附加策略。

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

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

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. 在 /flink 上打开适用于 Apache Flink 的托管服务控制台 https://console.aws.amazon.com

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

3. 选择 配置。

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

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

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

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

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

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

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

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

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

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

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

4. 如果您为 Firehose 直播的目标创建了 Amazon S3 存储桶，请同时删除该存储桶。

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

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

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

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

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

6. 如果您为 Firehose 直播创建了新政策，请同时删除该政策。

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

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

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

10. 如果您为 Firehose 直播创建了新角色，请同时删除该角色。

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

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

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

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

1. 编辑入门教程中的 kinesis-analytics-service-MyApplication-us-west-2 策略。该策略允许担任源账户中的角色，以便读取源流。

   注意

   当您使用控制台创建应用程序时，控制台会创建一个名为 kinesis-analytics-service-<application name>-<application region> 的策略和一个名为 kinesisanalytics-<application name>-<application region> 的角色。

   将下面突出显示的部分添加到策略中。将示例账户 ID (SOURCE01234567) 替换为您要用于源直播的账户 ID。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "AssumeRoleInSourceAccount",
               "Effect": "Allow",
               "Action": "sts:AssumeRole",
               "Resource": "arn:aws:iam::SOURCE01234567:role/KA-Source-Stream-Role"
           },
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:SINK012345678:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:SINK012345678:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:SINK012345678:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           }
       ]
   }
   

2. 打开 kinesis-analytics-MyApplication-us-west-2 角色，并记下其 Amazon 资源名称 (ARN)。您需要在下一部分中使用该名称。角色 ARN 如下所示。

   arn:aws:iam::SINK012345678:role/service-role/kinesis-analytics-MyApplication-us-west-2