为 Python 应用程序创建和运行 Kinesis Data Analytics

创建数据流 (Amazon CLI)

1. 创建第一个流 (ExampleInputStream)，请使用以下 Amazon Kinesiscreate-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
   import time
   
   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")
           time.sleep(2)
   
   
   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-java-examples

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

使用控制台上传应用程序代码：

1. 使用首选的压缩应用程序来压缩getting-started.py和https://mvnrepository.com/artifact/org.apache.flink/flink-sql 连接器-kinesis_2.12/1.13.2文件。命名存档myapp.zip. 如果在存档中包含外层文件夹，则必须将其与配置文件中的代码一起包含在路径中：GettingStarted/getting-started.py.

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

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

4. Enterka-app-code-<username>中的Bucket name字段中返回的子位置类型。将后缀（如您的用户名）添加到存储桶名称，以使其具有全局唯一性。选择 Next（下一步）。

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

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

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

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

9. 在选择文件步骤中，选择添加文件。导航到您在上一步中创建的 myapp.zip 文件。选择 Next（下一步）。

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

使用Amazon CLI：

1. 使用首选的压缩应用程序来压缩getting-started.py和https://mvnrepository.com/artifact/org.apache.flink/flink-sql 连接器-kinesis_2.12/1.13.2文件。命名存档myapp.zip. 如果在存档中包含外层文件夹，则必须将其与配置文件中的代码一起包含在路径中：GettingStarted/getting-started.py.

2. 运行以下 命令：

   $ aws s3 --region aws region cp myapp.zip s3://ka-app-code-<username>

1. 打开 Kinesis Data Analytics 控制台https://console.aws.amazon.com/kinesisanalytics.

2. 在 Kinesis Data Analytics 仪表板上，选择创建分析应用.

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

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

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

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

   • 将版本保留为Apache Flink 版本 1.13.2（推荐版本）.

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

5. 选择 Create application（创建应用程序）。

配置应用程序

1. 在 MyApplication (我的应用程序) 页面上，选择 Configure (配置)。

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

   • 适用于Amazon S3 存储桶输入ka-app-code-<username>.

   • 适用于Amazon S3 对象的路径输入myapp.zip.

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

4. UNDER属性，选择添加组. 适用于Group ID (组 ID)输入consumer.config.0.

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

   密钥	值
   input.stream.name	ExampleInputStream
   aws.region	us-west-2
   flink.stream.initpos	LATEST

   选择 Save（保存）。

6. UNDER属性，选择添加组。适用于Group ID (组 ID)输入producer.config.0.

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

   密钥	值
   output.stream.name	ExampleOutputStream
   aws.region	us-west-2
   shard.count	1

8. UNDER属性，选择添加组。适用于Group ID (组 ID)输入kinesis.analytics.flink.run.options. 这个特殊的属性组告诉你的应用程序在哪里找到其代码资源。有关更多信息，请参阅 指定代码文件。

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

   密钥	值
   python	getting-started.py
   jarfile	flink-sql-connector-kinesis_2.12-1.13.2.jar

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

11. 对于 CloudWatch logging (CloudWatch 日志记录)，请选中 Enable (启用) 复选框。

12. 选择 Update（更新）。

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

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

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

3. 在 Summary (摘要) 页面上，选择 Edit policy (编辑策略)。请选择 JSON 选项卡。

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

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/myapp.zip"
               ]
           },
           {
               "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"
           }
       ]
   }