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

1. 确保安装了 Apache Maven，并且您的JAVA_HOME环境变量指向 a 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参数。

报名参加 AWS 账户

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

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

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

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

保护你的 AWS 账户根用户

1. 选择 Root 用户并输入您的 AWS 账户 电子邮件地址，以账户所有者的身份登录。AWS Management Console在下一页上，输入您的密码。

   要获取使用根用户登录方面的帮助，请参阅《AWS 登录 用户指南》中的以根用户身份登录。

2. 为您的 root 用户开启多重身份验证 (MFA)。

   有关说明，请参阅《用户指南》中的 “为 AWS 账户 root 用户（控制台）启用虚拟MFA设备” IAM。

创建具有管理访问权限的用户

1. 启用IAM身份中心。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的启用 AWS IAM Identity Center。

2. 在 IAM Identity Center 中，向用户授予管理访问权限。

   有关使用 IAM Identity Center 目录 作为身份源的教程，请参阅《用户指南》 IAM Identity Center 目录中的使用默认设置配置AWS IAM Identity Center 用户访问权限。

以具有管理访问权限的用户身份登录

• 要使用您的 Ident IAM ity Center 用户登录URL，请使用您在创建 Ident IAM ity Center 用户时发送到您的电子邮件地址的登录信息。

  有关使用 Ident IAM ity Center 用户登录的帮助，请参阅AWS 登录 用户指南中的登录 AWS 访问门户。

将访问权限分配给其他用户

1. 在 IAM Identity Center 中，创建一个遵循应用最低权限权限的最佳实践的权限集。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的创建权限集。

2. 将用户分配到一个组，然后为该组分配单点登录访问权限。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的添加组。

要设置 AWS CLI

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

   • 安装 AWS Command Line Interface

   • 配置 AWS CLI

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

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

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

   注意

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

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

   aws help

创建数据流 (AWS CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesis 命令create-stream AWS 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 工具。通过在包含该JAR文件的目录中运行以下命令来创建您的pom.xml文件：

     mvn package -Dflink.version=1.13.2

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

     注意

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

   您可以将包作为JAR文件上传，也可以压缩包并将其作为ZIP文件上传。如果您使用创建应用程序 AWS 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/iam/。

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

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

   选择下一步: 权限。

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

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

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

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

   注意

   对于本练习，Managed Service for 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 选择 删除日志组，然后确认删除。

报名参加 AWS 账户

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

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

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

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

保护你的 AWS 账户根用户

1. 选择 Root 用户并输入您的 AWS 账户 电子邮件地址，以账户所有者的身份登录。AWS Management Console在下一页上，输入您的密码。

   要获取使用根用户登录方面的帮助，请参阅《AWS 登录 用户指南》中的以根用户身份登录。

2. 为您的 root 用户开启多重身份验证 (MFA)。

   有关说明，请参阅《用户指南》中的 “为 AWS 账户 root 用户（控制台）启用虚拟MFA设备” IAM。

创建具有管理访问权限的用户

1. 启用IAM身份中心。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的启用 AWS IAM Identity Center。

2. 在 IAM Identity Center 中，向用户授予管理访问权限。

   有关使用 IAM Identity Center 目录 作为身份源的教程，请参阅《用户指南》 IAM Identity Center 目录中的使用默认设置配置AWS IAM Identity Center 用户访问权限。

以具有管理访问权限的用户身份登录

• 要使用您的 Ident IAM ity Center 用户登录URL，请使用您在创建 Ident IAM ity Center 用户时发送到您的电子邮件地址的登录信息。

  有关使用 Ident IAM ity Center 用户登录的帮助，请参阅AWS 登录 用户指南中的登录 AWS 访问门户。

将访问权限分配给其他用户

1. 在 IAM Identity Center 中，创建一个遵循应用最低权限权限的最佳实践的权限集。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的创建权限集。

2. 将用户分配到一个组，然后为该组分配单点登录访问权限。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的添加组。

要设置 AWS CLI

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

   • 安装 AWS Command Line Interface

   • 配置 AWS CLI

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

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

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

   注意

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

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

   aws help

创建数据流 (AWS CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesis 命令create-stream AWS 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 工具。通过在包含该JAR文件的目录中运行以下命令来创建您的pom.xml文件：

     mvn package -Dflink.version=1.11.3

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

     注意

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

   您可以将包作为JAR文件上传，也可以压缩包并将其作为ZIP文件上传。如果您使用创建应用程序 AWS 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/iam/。

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

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

   选择下一步: 权限。

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

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

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

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

   注意

   对于本练习，Managed Service for 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 选择 删除日志组，然后确认删除。

报名参加 AWS 账户

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

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

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

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

保护你的 AWS 账户根用户

1. 选择 Root 用户并输入您的 AWS 账户 电子邮件地址，以账户所有者的身份登录。AWS Management Console在下一页上，输入您的密码。

   要获取使用根用户登录方面的帮助，请参阅《AWS 登录 用户指南》中的以根用户身份登录。

2. 为您的 root 用户开启多重身份验证 (MFA)。

   有关说明，请参阅《用户指南》中的 “为 AWS 账户 root 用户（控制台）启用虚拟MFA设备” IAM。

创建具有管理访问权限的用户

1. 启用IAM身份中心。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的启用 AWS IAM Identity Center。

2. 在 IAM Identity Center 中，向用户授予管理访问权限。

   有关使用 IAM Identity Center 目录 作为身份源的教程，请参阅《用户指南》 IAM Identity Center 目录中的使用默认设置配置AWS IAM Identity Center 用户访问权限。

以具有管理访问权限的用户身份登录

• 要使用您的 Ident IAM ity Center 用户登录URL，请使用您在创建 Ident IAM ity Center 用户时发送到您的电子邮件地址的登录信息。

  有关使用 Ident IAM ity Center 用户登录的帮助，请参阅AWS 登录 用户指南中的登录 AWS 访问门户。

将访问权限分配给其他用户

1. 在 IAM Identity Center 中，创建一个遵循应用最低权限权限的最佳实践的权限集。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的创建权限集。

2. 将用户分配到一个组，然后为该组分配单点登录访问权限。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的添加组。

要设置 AWS CLI

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

   • 安装 AWS Command Line Interface

   • 配置 AWS CLI

2. 在文件中为管理员用户添加已命名的配置 AWS CLI config文件。在执行 AWS CLI 命令时，您将使用此配置文件。有关命名配置文件的更多信息，请参阅 AWS 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 一般参考中的区域和终端节点。

   注意

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

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

   aws help

创建数据流 (AWS CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesis 命令create-stream AWS 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 工具。通过在包含该JAR文件的目录中运行以下命令来创建您的pom.xml文件：

     mvn package -Dflink.version=1.8.2

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

     注意

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

   您可以将包作为JAR文件上传，也可以压缩包并将其作为ZIP文件上传。如果您使用创建应用程序 AWS 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/iam/。

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

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

   选择下一步: 权限。

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

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

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

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

   注意

   对于本练习，Managed Service for 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 选择 删除日志组，然后确认删除。

报名参加 AWS 账户

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

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

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

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

保护你的 AWS 账户根用户

1. 选择 Root 用户并输入您的 AWS 账户 电子邮件地址，以账户所有者的身份登录。AWS Management Console在下一页上，输入您的密码。

   要获取使用根用户登录方面的帮助，请参阅《AWS 登录 用户指南》中的以根用户身份登录。

2. 为您的 root 用户开启多重身份验证 (MFA)。

   有关说明，请参阅《用户指南》中的 “为 AWS 账户 root 用户（控制台）启用虚拟MFA设备” IAM。

创建具有管理访问权限的用户

1. 启用IAM身份中心。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的启用 AWS IAM Identity Center。

2. 在 IAM Identity Center 中，向用户授予管理访问权限。

   有关使用 IAM Identity Center 目录 作为身份源的教程，请参阅《用户指南》 IAM Identity Center 目录中的使用默认设置配置AWS IAM Identity Center 用户访问权限。

以具有管理访问权限的用户身份登录

• 要使用您的 Ident IAM ity Center 用户登录URL，请使用您在创建 Ident IAM ity Center 用户时发送到您的电子邮件地址的登录信息。

  有关使用 Ident IAM ity Center 用户登录的帮助，请参阅AWS 登录 用户指南中的登录 AWS 访问门户。

将访问权限分配给其他用户

1. 在 IAM Identity Center 中，创建一个遵循应用最低权限权限的最佳实践的权限集。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的创建权限集。

2. 将用户分配到一个组，然后为该组分配单点登录访问权限。

   有关说明，请参阅《AWS IAM Identity Center 用户指南》中的添加组。

要设置 AWS CLI

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

   • 安装 AWS Command Line Interface

   • 配置 AWS CLI

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

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

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

   注意

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

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

   aws help

创建数据流 (AWS CLI)

1. 要创建第一个直播 (ExampleInputStream)，请使用以下 Amazon Kinesis 命令create-stream AWS 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 工具。通过在包含该JAR文件的目录中运行以下命令来创建您的pom.xml文件：

     mvn package

     注意

     Managed Service for Apache Flink 运行时系统版本 1.0.1 不需要 -Dflink.version 参数；仅 1.1.0 及更高版本需要使用该参数。有关更多信息，请参阅 指定应用程序的 Apache Flink 版本。

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

   您可以将包作为JAR文件上传，也可以压缩包并将其作为ZIP文件上传。如果您使用创建应用程序 AWS 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/iam/。

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

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

   选择下一步: 权限。

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

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

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

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

   注意

   对于本练习，Managed Service for 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 选择创建应用程序。

   注意

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

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

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

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

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

7. 选择创建应用程序。

1. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/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 Logs 或 AmazonVPC）时，控制台会修改您的应用程序执行角色以授予访问这些资源的权限。

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

   组 ID	键	值
   KafkaSource	topic	AWS 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	AWS 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

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

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

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

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

8. 在虚拟私有云 (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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/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>存储桶，然后选择上传。

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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 打开IAM控制台，网址为https://console.aws.amazon.com/iam/。

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

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

   选择下一步: 权限。

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

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

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

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

   注意

   对于本练习，Managed Service for 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. 打开IAM控制台，网址为https://console.aws.amazon.com/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. 选择 删除日志组，然后确认删除。