创建并运行 Managed Service for Apache Flink 应用程序

1. 连接到您在使用 Amazon MSK 入门教程的步骤 4：创建客户端计算机中创建的客户端实例。

2. 安装 Python3、Pip 和 Kafka Python 库：

   $ sudo yum install python37
   $ curl -O https://bootstrap.pypa.io/get-pip.py
   $ python3 get-pip.py --user
   $ pip install kafka-python
   

3. 使用以下内容创建名为 stock.py 的文件。将该BROKERS值替换为您之前记录的引导代理列表。

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

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

   $ python3 stock.py

下载 Java 应用程序代码

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

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

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

编译应用程序代码

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

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

     mvn package -Dflink.version=1.15.3

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

     注意

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

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

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

上传应用程序代码

1. 通过 https://console.aws.amazon.com/s3/ 打开 Amazon 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. 打开 Managed Service for Apache Flink 控制台，网址为 https://console.aws.amazon.com/flink

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

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

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

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

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

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

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

5. 选择创建应用程序。

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

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

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

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

4. 将以下策略示例中突出显示的部分添加到策略中。

   {
       "Version": "2012-10-17",
       "Statement": [
           {
           "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": "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"
               ]
           }
        ]
   }

配置应用程序

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	键	值
   FlinkApplicationProperties	kafka-topic	AWSKafkaTutorialTopic
   FlinkApplicationProperties	brokers	Your Amazon MSK cluster's Bootstrap Brokers list
   FlinkApplicationProperties	s3Path	ka-app-code-<username>
   FlinkApplicationProperties	security.protocol	SSL
   FlinkApplicationProperties	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   FlinkApplicationProperties	ssl.truststore.password	changeit

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

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

8. 在虚拟私有云 (VPC) 部分，选择基于Amazon MSK 集群的 VPC 配置。选择AWSKafkaTutorialCluster。

9. 选择更新。

运行应用程序

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

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

3. 在您的 Amazon EC2 客户端上，运行您之前创建的 Python 脚本，将记录写入 Amazon MSK 集群供您的应用程序处理：

   $ python3 stock.py