The Java application code for this example is available from GitHub. To download the application code, do the following:
1. Install the Git client if you haven't already. For more information, see Installing Git.
2. Clone the remote repository with the following command:
git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-java-examples 3. Navigate to the amazon-kinesis-data-analytics-java-examples/TumblingWindow
directory.
The application code is located in the TumblingWindowStreamingJob.java file. Note the following about the application code:
• The application uses a Kinesis source to read from the source stream. The following snippet creates the Kinesis source:
return env.addSource(new FlinkKinesisConsumer<>(inputStreamName, new SimpleStringSchema(), inputProperties));
• Add the following import statement:
import
org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows; //
flink 1.13
• The application uses the timeWindow operator to find the count of values for each stock symbol over a 5-second tumbling window. The following code creates the operator and sends the aggregated data to a new Kinesis Data Streams sink:
input.flatMap(new Tokenizer()) // Tokenizer for generating words
.keyBy(0) // Logically partition the stream for each word
//.timeWindow(Time.seconds(5)) // Tumbling window definition (Flink 1.11) .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) //Flink 1.13
.sum(1) // Sum the number of words per partition
Tumbling Window
.map(value -> value.f0 + "," + value.f1.toString() + "\n") .addSink(createSinkFromStaticConfig());
Compile the Application Code
To compile the application, do the following:
1. Install Java and Maven if you haven't already. For more information, see Prerequisites (p. 77) in the Getting Started (DataStream API) (p. 77) tutorial.
2. Compile the application with the following command:
mvn package -Dflink.version=1.13.2
NoteThe provided source code relies on libraries from Java 11. If you are using a development environment,
Compiling the application creates the application JAR file (target/aws-kinesis-analytics-java-apps-1.0.jar).
Upload the Apache Flink Streaming Java Code
In this section, you upload your application code to the Amazon S3 bucket you created in the Create Dependent Resources (p. 116) section.
1. In the Amazon S3 console, choose the ka-app-code-<username> bucket, and choose Upload.
2. 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.
3. You don't need to change any of the settings for the object, so choose Upload.
Your application code is now stored in an Amazon S3 bucket where your application can access it.
Create and Run the Kinesis Data Analytics Application
Follow these steps to create, configure, update, and run the application using the console.
Create the Application
1. Open the Kinesis Data Analytics console at https://console.aws.amazon.com/kinesisanalytics.
2. On the Kinesis Data Analytics dashboard, choose Create analytics application.
3. On the Kinesis Analytics - Create application page, provide the application details as follows:
• For Application name, enter MyApplication.
• For Runtime, choose Apache Flink.
NoteKinesis Data Analytics uses Apache Flink version 1.13.2.
• Leave the version pulldown as Apache Flink version 1.13.2 (Recommended version).
4. For Access permissions, choose Create / update IAM role kinesis-analytics-MyApplication-us-west-2.
5. Choose Create application.
Tumbling Window
NoteWhen you create a Kinesis Data Analytics application using the console, you have the option of having an IAM role and policy created for your application. Your application uses this role and policy to access its dependent resources. These IAM resources are named using your application name and Region as follows:
• Policy: kinesis-analytics-service-MyApplication-us-west-2
• Role: kinesis-analytics-MyApplication-us-west-2
Edit the IAM Policy
Edit the IAM policy to add permissions to access the Kinesis data streams.
1. Open the IAM console at https://console.aws.amazon.com/iam/.
2. Choose Policies. Choose the kinesis-analytics-service-MyApplication-us-west-2 policy that the console created for you in the previous section.
3. On the Summary page, choose Edit policy. Choose the JSON tab.
4. Add the highlighted section of the following policy example to the policy. Replace the sample account IDs (012345678901) with your account ID.
{
"Version": "2012-10-17", "Statement": [
"arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
"Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
"Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
Tumbling Window {
"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"
} ] }
Configure the Application
1. On the MyApplication page, choose Configure.
2. On the Configure application page, provide the Code location:
• For Amazon S3 bucket, enter ka-app-code-<username>.
• For Path to Amazon S3 object, enter aws-kinesis-analytics-java-apps-1.0.jar.
3. Under Access to application resources, for Access permissions, choose Create / update IAM role kinesis-analytics-MyApplication-us-west-2.
4. Under Monitoring, ensure that the Monitoring metrics level is set to Application.
5. For CloudWatch logging, select the Enable check box.
6. Choose Update.
NoteWhen you choose to enable CloudWatch logging, Kinesis Data Analytics creates a log group and log stream for you. The names of these resources are as follows:
• Log group: /aws/kinesis-analytics/MyApplication
• Log stream: kinesis-analytics-log-stream
This log stream is used to monitor the application. This is not the same log stream that the application uses to send results.
Run the Application
1. On the MyApplication page, choose Run. Leave the Run without snapshot option selected, and confirm the action.
2. When the application is running, refresh the page. The console shows the Application graph.
You can check the Kinesis Data Analytics metrics on the CloudWatch console to verify that the application is working.
Clean Up AWS Resources
This section includes procedures for cleaning up AWS resources created in the Tumbling Window tutorial.
This topic contains the following sections:
Tumbling Window
• Delete Your Kinesis Data Analytics Application (p. 121)
• Delete Your Kinesis Data Streams (p. 121)
• Delete Your Amazon S3 Object and Bucket (p. 121)
• Delete Your IAM Resources (p. 121)
• Delete Your CloudWatch Resources (p. 121)
Delete Your Kinesis Data Analytics Application
1. Open the Kinesis Data Analytics console at https://console.aws.amazon.com/kinesisanalytics.
2. in the Kinesis Data Analytics panel, choose MyApplication.
3. In the application's page, choose Delete and then confirm the deletion.
Delete Your Kinesis Data Streams
1. Open the Kinesis console at https://console.aws.amazon.com/kinesis.
2. In the Kinesis Data Streams panel, choose ExampleInputStream.
3. In the ExampleInputStream page, choose Delete Kinesis Stream and then confirm the deletion.
4. In the Kinesis streams page, choose the ExampleOutputStream, choose Actions, choose Delete, and then confirm the deletion.
Delete Your Amazon S3 Object and Bucket
1. Open the Amazon S3 console at https://console.aws.amazon.com/s3/.
2. Choose the ka-app-code-<username> bucket.
3. Choose Delete and then enter the bucket name to confirm deletion.
Delete Your IAM Resources
1. Open the IAM console at https://console.aws.amazon.com/iam/.
2. In the navigation bar, choose Policies.
3. In the filter control, enter kinesis.
4. Choose the kinesis-analytics-service-MyApplication-<your-region> policy.
5. Choose Policy Actions and then choose Delete.
6. In the navigation bar, choose Roles.
7. Choose the kinesis-analytics-MyApplication-<your-region> role.
8. Choose Delete role and then confirm the deletion.
Delete Your CloudWatch Resources
1. Open the CloudWatch console at https://console.aws.amazon.com/cloudwatch/.
2. In the navigation bar, choose Logs.
3. Choose the /aws/kinesis-analytics/MyApplication log group.
4. Choose Delete Log Group and then confirm the deletion.
Sliding Window
Example: Sliding Window
In this exercise, you create a Kinesis Data Analytics application that aggregates data using a sliding window.
Note
To set up required prerequisites for this exercise, first complete the Getting Started (DataStream API) (p. 77) exercise.
This topic contains the following sections:
• Create Dependent Resources (p. 122)
• Write Sample Records to the Input Stream (p. 122)
• Download and Examine the Application Code (p. 123)
• Compile the Application Code (p. 124)
• Upload the Apache Flink Streaming Java Code (p. 124)
• Create and Run the Kinesis Data Analytics Application (p. 124)
• Clean Up AWS Resources (p. 127)
Create Dependent Resources
Before you create a Kinesis Data Analytics application for this exercise, you create the following dependent resources:
• Two Kinesis data streams (ExampleInputStream and ExampleOutputStream).
• An Amazon S3 bucket to store the application's code (ka-app-code-<username>)
You can create the Kinesis streams and Amazon S3 bucket using the console. For instructions for creating these resources, see the following topics:
• Creating and Updating Data Streams in the Amazon Kinesis Data Streams Developer Guide. Name your data streams ExampleInputStream and ExampleOutputStream.
• How Do I Create an S3 Bucket? in the Amazon Simple Storage Service User Guide. Give the Amazon S3 bucket a globally unique name by appending your login name, such as ka-app-code-<username>.
Write Sample Records to the Input Stream
In this section, you use a Python script to write sample records to the stream for the application to process.
Note
This section requires the AWS SDK for Python (Boto).
1. Create a file named stock.py with the following contents:
import datetime import json import random import boto3
STREAM_NAME = "ExampleInputStream"
Sliding Window
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. Run the stock.py script:
$ python stock.py
Keep the script running while completing the rest of the tutorial.
Download and Examine the Application Code
The Java application code for this example is available from GitHub. To download the application code, do the following:
1. Install the Git client if you haven't already. For more information, see Installing Git.
2. Clone the remote repository with the following command:
git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-java-examples 3. Navigate to the amazon-kinesis-data-analytics-java-examples/SlidingWindow
directory.
The application code is located in the SlidingWindowStreamingJobWithParallelism.java file.
Note the following about the application code:
• The application uses a Kinesis source to read from the source stream. The following snippet creates the Kinesis source:
return env.addSource(new FlinkKinesisConsumer<>(inputStreamName, new SimpleStringSchema(), inputProperties));
• The application uses the timeWindow operator to find the minimum value for each stock symbol over a 10-second window that slides by 5 seconds. The following code creates the operator and sends the aggregated data to a new Kinesis Data Streams sink:
• Add the following import statement:
import
org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows; //
flink 1.13
Sliding Window
• The application uses the timeWindow operator to find the count of values for each stock symbol over a 5-second tumbling window. The following code creates the operator and sends the aggregated data to a new Kinesis Data Streams sink:
input.flatMap(new Tokenizer()) // Tokenizer for generating words
.keyBy(0) // Logically partition the stream for each word
//.timeWindow(Time.seconds(5)) // Tumbling window definition (Flink 1.11) .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) //Flink 1.13
.sum(1) // Sum the number of words per partition
.map(value -> value.f0 + "," + value.f1.toString() + "\n") .addSink(createSinkFromStaticConfig());
Compile the Application Code
To compile the application, do the following:
1. Install Java and Maven if you haven't already. For more information, see Prerequisites (p. 77) in the Getting Started (DataStream API) (p. 77) tutorial.
2. Compile the application with the following command:
mvn package -Dflink.version=1.13.2
NoteThe provided source code relies on libraries from Java 11. If you are using a development environment,
Compiling the application creates the application JAR file (target/aws-kinesis-analytics-java-apps-1.0.jar).
Upload the Apache Flink Streaming Java Code
In this section, you upload your application code to the Amazon S3 bucket that you created in the Create Dependent Resources (p. 122) section.
1. In the Amazon S3 console, choose the ka-app-code-<username> bucket, and then choose Upload.
2. 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.
3. You don't need to change any of the settings for the object, so choose Upload.
Your application code is now stored in an Amazon S3 bucket where your application can access it.
Create and Run the Kinesis Data Analytics Application
Follow these steps to create, configure, update, and run the application using the console.
Create the Application
1. Open the Kinesis Data Analytics console at https://console.aws.amazon.com/kinesisanalytics.
2. On the Kinesis Data Analytics dashboard, choose Create analytics application.
3. On the Kinesis Analytics - Create application page, provide the application details as follows:
• For Application name, enter MyApplication.
Sliding Window
• For Runtime, choose Apache Flink.
• Leave the version pulldown as Apache Flink version 1.13.2 (Recommended version).
4. For Access permissions, choose Create / update IAM role kinesis-analytics-MyApplication-us-west-2.
5. Choose Create application.
NoteWhen you create a Kinesis Data Analytics application using the console, you have the option of having an IAM role and policy created for your application. Your application uses this role and policy to access its dependent resources. These IAM resources are named using your application name and Region as follows:
• Policy: kinesis-analytics-service-MyApplication-us-west-2
• Role: kinesis-analytics-MyApplication-us-west-2
Edit the IAM Policy
Edit the IAM policy to add permissions to access the Kinesis data streams.
1. Open the IAM console at https://console.aws.amazon.com/iam/.
2. Choose Policies. Choose the kinesis-analytics-service-MyApplication-us-west-2 policy that the console created for you in the previous section.
3. On the Summary page, choose Edit policy. Choose the JSON tab.
4. Add the highlighted section of the following policy example to the policy. Replace the sample account IDs (012345678901) with your account ID.
{ "Version": "2012-10-17", "Statement": [
"arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
"Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
"Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
}, {
Sliding Window "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"
} ] }