Example: Creating a Sliding Window in Scala

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/scala/SlidingWindow directory.

Compile the Application Code

Use the SBT build tool to build the Scala code for the application. To install SBT, see Install sbt with cs setup. You also need to install the Java Development Kit (JDK). See Prerequisites for Completing the Exercises.

1. To use your application code, you compile and package it into a JAR file. You can compile and package your code with SBT:

   sbt assembly

2. If the application compiles successfully, the following file is created:

   target/scala-3.2.0/sliding-window-scala-1.0.jar

Upload the Apache Flink Streaming Scala Code

In this section, you create an Amazon S3 bucket and upload your application code.

1. Open the Amazon S3 console at https://console.aws.amazon.com/s3/.

2. Choose Create bucket

3. Enter ka-app-code-<username> in the Bucket name field. Add a suffix to the bucket name, such as your user name, to make it globally unique. Choose Next.

4. In Configure options, keep the settings as they are, and choose Next.

5. In Set permissions, keep the settings as they are, and choose Next.

6. Choose Create bucket.

7. Choose the ka-app-code-<username> bucket, and then choose Upload.

8. In the Select files step, choose Add files. Navigate to the sliding-window-scala-1.0.jar file that you created in the previous step.

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

1. Open the Kinesis Data Analytics console at https://console.amazonaws.cn/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 Description, enter My Scala test app.

   • For Runtime, choose Apache Flink.

   • Leave the version as Apache Flink version 1.15.2 (Recommended version).

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

5. Choose Create application.

To 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 sliding-window-scala-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 Properties, choose Add group.

5. Enter the following:

   Group ID	Key	Value
   ConsumerConfigProperties	input.stream.name	ExampleInputStream
   ConsumerConfigProperties	aws.region	us-west-2
   ConsumerConfigProperties	flink.stream.initpos	LATEST

   Choose Save.

6. Under Properties, choose Add group again.

7. Enter the following:

   Group ID	Key	Value
   ProducerConfigProperties	output.stream.name	ExampleOutputStream
   ProducerConfigProperties	aws.region	us-west-2

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

9. For CloudWatch logging, choose the Enable check box.

10. Choose Update.

To edit the IAM policy to add S3 bucket permissions

1. Open the IAM console at https://console.amazonaws.cn/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": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/sliding-window-scala-1.0.jar"
               ]
           },
           {
               "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"
           }
       ]
   }

To create an IAM role

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

2. In the navigation pane, choose Roles and then Create Role.

3. Under Select type of trusted identity, choose Amazon Service

4. Under Choose the service that will use this role, choose Kinesis.

5. Under Select your use case, choose Kinesis Analytics.

6. Choose Next: Permissions.

7. On the Attach permissions policies page, choose Next: Review. You attach permissions policies after you create the role.

8. On the Create role page, enter KA-stream-rw-role for the Role name. Choose Create role.

   Now you have created a new IAM role called KA-stream-rw-role. Next, you update the trust and permissions policies for the role

9. Attach the permissions policy to the role.

   Note

   For this exercise, Kinesis Data Analytics assumes this role for both reading data from a Kinesis data stream (source) and writing output to another Kinesis data stream. So you attach the policy that you created in the previous step, Create a Permissions Policy.

   1. On the Summary page, choose the Permissions tab.

   2. Choose Attach Policies.

   3. In the search box, enter KAReadSourceStreamWriteSinkStream (the policy that you created in the previous section).

   4. Choose the KAReadSourceStreamWriteSinkStream policy, and choose Attach policy.

To start the application

1. Save the following JSON code to a file named start_request.json.

   {
       "ApplicationName": "sliding_window",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }

2. Execute the StartApplication action with the preceding request to start the application:

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

To stop the application

1. Save the following JSON code to a file named stop_request.json.

   {
      "ApplicationName": "sliding_window"
   }

2. Execute the StopApplication action with the preceding request to stop the application:

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

To update environment properties for the application

1. Save the following JSON code to a file named update_properties_request.json.

   {"ApplicationName": "sliding_window",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2",
                       "stream.name" : "ExampleInputStream",
                       "flink.stream.initpos" : "LATEST"
                  }
               },
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2",
                       "stream.name" : "ExampleOutputStream"
                  }
               }
            ]
         }
      }
   }

2. Execute the UpdateApplication action with the preceding request to update environment properties:

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. in the Kinesis Data Analytics panel, choose MyApplication.

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

1. Open the Kinesis console at https://console.amazonaws.cn/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.

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

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

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

1. Open the IAM console at https://console.amazonaws.cn/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.

1. Open the CloudWatch console at https://console.amazonaws.cn/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.