示例：写入 Amazon S3 存储桶

使用以下内容创建名为 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'))

运行 stock.py 脚本：

$ python stock.py

在完成本教程的其余部分时，请将脚本保持运行状态。

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

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

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

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

如果还没有 Java 和 Maven，请安装它们。有关更多信息，请参阅入门 (DataStreamAPI)教程中的先决条件。

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

mvn package -Dflink.version=1.13.2

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

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

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

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

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

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

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

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

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

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

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

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

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

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

{
            "Sid": "ReadCode",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:GetObjectVersion"
            ],
            "Resource": [
                "arn:aws:s3:::kinesis-analytics-placeholder-s3-bucket/kinesis-analytics-placeholder-s3-object"
            ]
        },
        {
            "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"
        },
        {
            "Sid": "WriteObjects",
            "Effect": "Allow",
            "Action": [
                "s3:Abort*",
                "s3:DeleteObject*",
                "s3:GetObject*",
                "s3:GetBucket*",
                "s3:List*",
                "s3:ListBucket",
                "s3:PutObject"
            ],
            "Resource": [
                "arn:aws:s3:::ka-app-<username>",
                "arn:aws:s3:::ka-app-<username>/*"
            ]
        }
    ]
}

在存储库的MyApplication页面上，选择配置.

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

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

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

适用于CloudWatch 记录，选择启用”复选框。

选择 Update（更新）。

在存储库的MyApplication页面上，选择运行. 离开了在没有快照的情况下运行选项，然后确认操作。

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

将以下导入语句添加至开头S3StreamingSinkJob.javafile:

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

更新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;
}
            

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

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

要使用上述代码，请指定当前应用程序版本。您可以使用ListApplicationsaction.

将以下导入语句添加至开头S3StreamingSinkJob.javafile:

import java.util.concurrent.TimeUnit; 

更新createS3SinkFromStaticConfig中的方法S3StreamingSinkJob.java文件与以下内容类似：

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 分钟。

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

在 Kinesis Data Analytics 面板中，选择MyApplication.

在应用程序页面上，选择Delete然后确认删除。

打开 Kinesis 控制台，网址为：https://console.aws.amazon.com/kinesis。

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

在存储库的ExampleInputStream页面上，选择删除 Kinesis Streams然后确认删除。

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

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

选择 Delete (删除)，然后输入存储桶名称以确认删除。

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

在导航栏上，选择策略.

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

选择kinesis-analytics-service-MyApplication-<your-region>政策。

选择 Policy Actions (策略操作)，然后选择 Delete (删除)。

在导航栏上，选择角色.

选择kinesines-anticticMyApplication-<your-region>角色。

选择 Delete role (删除角色)，然后确认删除。

打开 CloudWatch 控制台https://console.aws.amazon.com/cloudwatch/.

在导航栏上，选择日志.

选择/aws/kinesis-analytictics/MyApplication日志组。

选择 Delete Log Group (删除日志组)，然后确认删除。