Example: Writing to an Amazon S3 Bucket

Create a file named stock.py with the following contents:

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

Run the stock.py script:

$ python stock.py

Keep the script running while completing the rest of the tutorial.

Install the Git client if you haven't already. For more information, see Installing Git.

Clone the remote repository with the following command:

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

Navigate to the amazon-kinesis-data-analytics-java-examples/S3Sink directory.

Install Java and Maven if you haven't already. For more information, see Prerequisites in the Getting Started (DataStream API) tutorial.

Compile the application with the following command:

mvn package -Dflink.version=1.15.3

In the Amazon S3 console, choose the ka-app-code-<username> bucket, navigate to the code folder, and choose Upload.

In the Select files step, choose Add files. Navigate to the aws-kinesis-analytics-java-apps-1.0.jar file that you created in the previous step.

You don't need to change any of the settings for the object, so choose Upload.

Open the Kinesis Data Analytics console at https://console.amazonaws.cn/kinesisanalytics.

On the Kinesis Data Analytics dashboard, choose Create analytics application.

On the Kinesis Analytics - Create application page, provide the application details as follows:

For Access permissions, choose Create / update IAM role kinesis-analytics-MyApplication-us-west-2.

Choose Create application.

For Access permissions, choose Create / update IAM role kinesis-analytics-MyApplication-us-west-2.

Choose Create application.

Open the IAM console at https://console.amazonaws.cn/iam/.

Choose Policies. Choose the kinesis-analytics-service-MyApplication-us-west-2 policy that the console created for you in the previous section.

On the Summary page, choose Edit policy. Choose the JSON tab.

Add the highlighted section of the following policy example to the policy. Replace the sample account IDs (012345678901) with your account ID. Replace <username> with your user name.

{
            "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"
        },
        
    ]
}

On the MyApplication page, choose Configure.

On the Configure application page, provide the Code location:

Under Access to application resources, for Access permissions, choose Create / update IAM role kinesis-analytics-MyApplication-us-west-2.

Under Monitoring, ensure that the Monitoring metrics level is set to Application.

For CloudWatch logging, select the Enable check box.

Choose Update.

On the MyApplication page, choose Run. Leave the Run without snapshot option selected, and confirm the action.

When the application is running, refresh the page. The console shows the Application graph.

Add the following import statements to the beginning of the S3StreamingSinkJob.java file:

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

Update the createS3SinkFromStaticConfig() method in the code to look like the following:

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

Increase the application's CheckpointInterval setting. The following input for the UpdateApplication action sets the checkpoint interval to 10 minutes:

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

To use the preceding code, specify the current application version. You can retrieve the application version by using the ListApplications action.

Add the following import statement to the beginning of the S3StreamingSinkJob.java file:

import java.util.concurrent.TimeUnit; 

Update the createS3SinkFromStaticConfig method in the S3StreamingSinkJob.java file to look like the following:

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

The preceding code example sets the frequency of writes to the Amazon S3 bucket to 8 minutes.

Open the Kinesis Data Analytics console at https://console.amazonaws.cn/kinesisanalytics.

In the Kinesis Data Analytics panel, choose MyApplication.

On the application's page, choose Delete and then confirm the deletion.

Open the Kinesis console at https://console.amazonaws.cn/kinesis.

In the Kinesis Data Streams panel, choose ExampleInputStream.

On the ExampleInputStream page, choose Delete Kinesis Stream and then confirm the deletion.

Open the Amazon S3 console at https://console.amazonaws.cn/s3/.

Choose the ka-app-code-<username> bucket.

Choose Delete and then enter the bucket name to confirm deletion.

Open the IAM console at https://console.amazonaws.cn/iam/.

On the navigation bar, choose Policies.

In the filter control, enter kinesis.

Choose the kinesis-analytics-service-MyApplication-<your-region> policy.

Choose Policy Actions and then choose Delete.

On the navigation bar, choose Roles.

Choose the kinesis-analytics-MyApplication-<your-region> role.

Choose Delete role and then confirm the deletion.

Open the CloudWatch console at https://console.amazonaws.cn/cloudwatch/.

On the navigation bar, choose Logs.

Choose the /aws/kinesis-analytics/MyApplication log group.

Choose Delete Log Group and then confirm the deletion.