Earlier version information for Managed Service for Apache Flink

1. Ensure that you have Apache Maven installed, and your JAVA_HOME environment variable points to a JDK rather than a JRE. You can test your Apache Maven install with the following command:

   mvn -version

2. Download the Apache Flink version 1.8.2 source code:

   wget https://archive.apache.org/dist/flink/flink-1.8.2/flink-1.8.2-src.tgz

3. Uncompress the Apache Flink source code:

   tar -xvf flink-1.8.2-src.tgz

4. Change to the Apache Flink source code directory:

   cd flink-1.8.2

5. Compile and install Apache Flink:

   mvn clean install -Pinclude-kinesis -DskipTests

   Note

   If you are compiling Flink on Microsoft Windows, you need to add the -Drat.skip=true parameter.

To sign up for an AWS account

1. Open https://portal.aws.amazon.com/billing/signup.

2. Follow the online instructions.

   Part of the sign-up procedure involves receiving a phone call and entering a verification code on the phone keypad.

   When you sign up for an AWS account, an AWS account root user is created. The root user has access to all AWS services and resources in the account. As a security best practice, assign administrative access to a user, and use only the root user to perform tasks that require root user access.

Secure your AWS account root user

1. Sign in to the AWS Management Console as the account owner by choosing Root user and entering your AWS account email address. On the next page, enter your password.

   For help signing in by using root user, see Signing in as the root user in the AWS Sign-In User Guide.

2. Turn on multi-factor authentication (MFA) for your root user.

   For instructions, see Enable a virtual MFA device for your AWS account root user (console) in the IAM User Guide.

Create a user with administrative access

1. Enable IAM Identity Center.

   For instructions, see Enabling AWS IAM Identity Center in the AWS IAM Identity Center User Guide.

2. In IAM Identity Center, grant administrative access to a user.

   For a tutorial about using the IAM Identity Center directory as your identity source, see Configure user access with the default IAM Identity Center directory in the AWS IAM Identity Center User Guide.

Sign in as the user with administrative access

• To sign in with your IAM Identity Center user, use the sign-in URL that was sent to your email address when you created the IAM Identity Center user.

  For help signing in using an IAM Identity Center user, see Signing in to the AWS access portal in the AWS Sign-In User Guide.

Assign access to additional users

1. In IAM Identity Center, create a permission set that follows the best practice of applying least-privilege permissions.

   For instructions, see Create a permission set in the AWS IAM Identity Center User Guide.

2. Assign users to a group, and then assign single sign-on access to the group.

   For instructions, see Add groups in the AWS IAM Identity Center User Guide.

To set up the AWS CLI

1. Download and configure the AWS CLI. For instructions, see the following topics in the AWS Command Line Interface User Guide:

   • Installing the AWS Command Line Interface

   • Configuring the AWS CLI

2. Add a named profile for the administrator user in the AWS CLI config file. You use this profile when executing the AWS CLI commands. For more information about named profiles, see Named Profiles in the AWS Command Line Interface User Guide.

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   For a list of available AWS Regions, see Regions and Endpoints in the Amazon Web Services General Reference.

   Note

   The example code and commands in this tutorial use the US West (Oregon) Region. To use a different Region, change the Region in the code and commands for this tutorial to the Region you want to use.

3. Verify the setup by entering the following help command at the command prompt:

   aws help

To create the data streams (AWS CLI)

1. To create the first stream (ExampleInputStream), use the following Amazon Kinesis create-stream AWS CLI command.

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

2. To create the second stream that the application uses to write output, run the same command, changing the stream name to ExampleOutputStream.

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

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

2. Later in the tutorial, you run the stock.py script to send data to the application.

   $ python stock.py

1. Clone the remote repository using the following command:

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

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

To compile the application code

1. To use your application code, you compile and package it into a JAR file. You can compile and package your code in one of two ways:

   • Use the command-line Maven tool. Create your JAR file by running the following command in the directory that contains the pom.xml file:

     mvn package -Dflink.version=1.13.2

   • Use your development environment. See your development environment documentation for details.

     Note

     The provided source code relies on libraries from Java 11.

   You can either upload your package as a JAR file, or you can compress your package and upload it as a ZIP file. If you create your application using the AWS CLI, you specify your code content type (JAR or ZIP).

2. If there are errors while compiling, verify that your JAVA_HOME environment variable is correctly set.

To upload the 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 the Configure options step, keep the settings as they are, and choose Next.

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

6. Choose Create bucket.

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

8. 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. Choose Next.

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Description, enter My java test app.

   • For Runtime, choose Apache Flink.

   • Leave the version pulldown as Apache Flink version 1.13.

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

5. Choose Create application.

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": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-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"
           }
       ]
   }

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. Enter the following:

   Group ID	Key	Value
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

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

6. For CloudWatch logging, select the Enable check box.

7. Choose Update.

To create an IAM role

1. Open the IAM console at https://console.aws.amazon.com/iam/.

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

3. Under Select type of trusted identity, choose AWS Service. Under Choose the service that will use this role, choose Kinesis. Under Select your use case, choose Kinesis Analytics.

   Choose Next: Permissions.

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

5. On the Create role page, enter MF-stream-rw-role for the Role name. Choose Create role.

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

6. Attach the permissions policy to the role.

   Note

   For this exercise, Managed Service for Apache Flink 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 AKReadSourceStreamWriteSinkStream (the policy that you created in the previous section).

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

1. Save the following JSON code to a file named create_request.json. Replace the sample role ARN with the ARN for the role that you created previously. Replace the bucket ARN suffix (username) with the suffix that you chose in the previous section. Replace the sample account ID (012345678901) in the service execution role with your account ID.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_15",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. Execute the CreateApplication action with the preceding request to create the application:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

To start the application

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

   {
       "ApplicationName": "test",
       "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": "test"
   }
   

2. Execute the StopApplication action with the following 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": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

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 console at https://console.aws.amazon.com/kinesis.

2. In the Managed Service for Apache Flink panel, choose MyApplication.

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

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.aws.amazon.com/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.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-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. In the navigation bar, choose Roles.

7. Choose the kinesis-analytics-MyApplication-us-west-2 role.

8. Choose Delete role and then confirm the deletion.

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.

To sign up for an AWS account

1. Open https://portal.aws.amazon.com/billing/signup.

2. Follow the online instructions.

   Part of the sign-up procedure involves receiving a phone call and entering a verification code on the phone keypad.

   When you sign up for an AWS account, an AWS account root user is created. The root user has access to all AWS services and resources in the account. As a security best practice, assign administrative access to a user, and use only the root user to perform tasks that require root user access.

Secure your AWS account root user

1. Sign in to the AWS Management Console as the account owner by choosing Root user and entering your AWS account email address. On the next page, enter your password.

   For help signing in by using root user, see Signing in as the root user in the AWS Sign-In User Guide.

2. Turn on multi-factor authentication (MFA) for your root user.

   For instructions, see Enable a virtual MFA device for your AWS account root user (console) in the IAM User Guide.

Create a user with administrative access

1. Enable IAM Identity Center.

   For instructions, see Enabling AWS IAM Identity Center in the AWS IAM Identity Center User Guide.

2. In IAM Identity Center, grant administrative access to a user.

   For a tutorial about using the IAM Identity Center directory as your identity source, see Configure user access with the default IAM Identity Center directory in the AWS IAM Identity Center User Guide.

Sign in as the user with administrative access

• To sign in with your IAM Identity Center user, use the sign-in URL that was sent to your email address when you created the IAM Identity Center user.

  For help signing in using an IAM Identity Center user, see Signing in to the AWS access portal in the AWS Sign-In User Guide.

Assign access to additional users

1. In IAM Identity Center, create a permission set that follows the best practice of applying least-privilege permissions.

   For instructions, see Create a permission set in the AWS IAM Identity Center User Guide.

2. Assign users to a group, and then assign single sign-on access to the group.

   For instructions, see Add groups in the AWS IAM Identity Center User Guide.

To set up the AWS CLI

1. Download and configure the AWS CLI. For instructions, see the following topics in the AWS Command Line Interface User Guide:

   • Installing the AWS Command Line Interface

   • Configuring the AWS CLI

2. Add a named profile for the administrator user in the AWS CLI config file. You use this profile when executing the AWS CLI commands. For more information about named profiles, see Named Profiles in the AWS Command Line Interface User Guide.

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   For a list of available AWS Regions, see Regions and Endpoints in the Amazon Web Services General Reference.

   Note

   The example code and commands in this tutorial use the US West (Oregon) Region. To use a different Region, change the Region in the code and commands for this tutorial to the Region you want to use.

3. Verify the setup by entering the following help command at the command prompt:

   aws help

To create the data streams (AWS CLI)

1. To create the first stream (ExampleInputStream), use the following Amazon Kinesis create-stream AWS CLI command.

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

2. To create the second stream that the application uses to write output, run the same command, changing the stream name to ExampleOutputStream.

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

1. 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"))
   

2. Later in the tutorial, you run the stock.py script to send data to the application.

   $ python stock.py

1. Clone the remote repository using the following command:

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

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

To compile the application code

1. To use your application code, you compile and package it into a JAR file. You can compile and package your code in one of two ways:

   • Use the command-line Maven tool. Create your JAR file by running the following command in the directory that contains the pom.xml file:

     mvn package -Dflink.version=1.11.3

   • Use your development environment. See your development environment documentation for details.

     Note

     The provided source code relies on libraries from Java 11. Ensure that your project's Java version is 11.

   You can either upload your package as a JAR file, or you can compress your package and upload it as a ZIP file. If you create your application using the AWS CLI, you specify your code content type (JAR or ZIP).

2. If there are errors while compiling, verify that your JAVA_HOME environment variable is correctly set.

To upload the 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 the Configure options step, keep the settings as they are, and choose Next.

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

6. Choose Create bucket.

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

8. 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. Choose Next.

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Description, enter My java test app.

   • For Runtime, choose Apache Flink.

   • Leave the version pulldown as Apache Flink version 1.11 (Recommended version).

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

5. Choose Create application.

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": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-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"
           }
       ]
   }

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 Properties, for Group ID, enter ProducerConfigProperties.

5. Enter the following application properties and values:

   Group ID	Key	Value
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

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

7. For CloudWatch logging, select the Enable check box.

8. Choose Update.

To create an IAM role

1. Open the IAM console at https://console.aws.amazon.com/iam/.

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

3. Under Select type of trusted identity, choose AWS Service. Under Choose the service that will use this role, choose Kinesis. Under Select your use case, choose Kinesis Analytics.

   Choose Next: Permissions.

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

5. On the Create role page, enter MF-stream-rw-role for the Role name. Choose Create role.

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

6. Attach the permissions policy to the role.

   Note

   For this exercise, Managed Service for Apache Flink 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 AKReadSourceStreamWriteSinkStream (the policy that you created in the previous section).

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

1. Save the following JSON code to a file named create_request.json. Replace the sample role ARN with the ARN for the role that you created previously. Replace the bucket ARN suffix (username) with the suffix that you chose in the previous section. Replace the sample account ID (012345678901) in the service execution role with your account ID.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_11",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. Execute the CreateApplication action with the preceding request to create the application:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

To start the application

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

   {
       "ApplicationName": "test",
       "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": "test"
   }
   

2. Execute the StopApplication action with the following 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": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

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 console at https://console.aws.amazon.com/kinesis.

2. In the Managed Service for Apache Flink panel, choose MyApplication.

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

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.aws.amazon.com/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.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-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. In the navigation bar, choose Roles.

7. Choose the kinesis-analytics-MyApplication-us-west-2 role.

8. Choose Delete role and then confirm the deletion.

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.

To sign up for an AWS account

1. Open https://portal.aws.amazon.com/billing/signup.

2. Follow the online instructions.

   Part of the sign-up procedure involves receiving a phone call and entering a verification code on the phone keypad.

   When you sign up for an AWS account, an AWS account root user is created. The root user has access to all AWS services and resources in the account. As a security best practice, assign administrative access to a user, and use only the root user to perform tasks that require root user access.

Secure your AWS account root user

1. Sign in to the AWS Management Console as the account owner by choosing Root user and entering your AWS account email address. On the next page, enter your password.

   For help signing in by using root user, see Signing in as the root user in the AWS Sign-In User Guide.

2. Turn on multi-factor authentication (MFA) for your root user.

   For instructions, see Enable a virtual MFA device for your AWS account root user (console) in the IAM User Guide.

Create a user with administrative access

1. Enable IAM Identity Center.

   For instructions, see Enabling AWS IAM Identity Center in the AWS IAM Identity Center User Guide.

2. In IAM Identity Center, grant administrative access to a user.

   For a tutorial about using the IAM Identity Center directory as your identity source, see Configure user access with the default IAM Identity Center directory in the AWS IAM Identity Center User Guide.

Sign in as the user with administrative access

• To sign in with your IAM Identity Center user, use the sign-in URL that was sent to your email address when you created the IAM Identity Center user.

  For help signing in using an IAM Identity Center user, see Signing in to the AWS access portal in the AWS Sign-In User Guide.

Assign access to additional users

1. In IAM Identity Center, create a permission set that follows the best practice of applying least-privilege permissions.

   For instructions, see Create a permission set in the AWS IAM Identity Center User Guide.

2. Assign users to a group, and then assign single sign-on access to the group.

   For instructions, see Add groups in the AWS IAM Identity Center User Guide.

To set up the AWS CLI

1. Download and configure the AWS CLI. For instructions, see the following topics in the AWS Command Line Interface User Guide:

   • Installing the AWS Command Line Interface

   • Configuring the AWS CLI

2. Add a named profile for the administrator user in the AWS CLI config file. You use this profile when executing the AWS CLI commands. For more information about named profiles, see Named Profiles in the AWS Command Line Interface User Guide.

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   For a list of available Regions, see Regions and Endpoints in the Amazon Web Services General Reference.

   Note

   The example code and commands in this tutorial use the US West (Oregon) Region. To use a different AWS Region, change the Region in the code and commands for this tutorial to the Region you want to use.

3. Verify the setup by entering the following help command at the command prompt:

   aws help

To create the data streams (AWS CLI)

1. To create the first stream (ExampleInputStream), use the following Amazon Kinesis create-stream AWS CLI command.

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

2. To create the second stream that the application uses to write output, run the same command, changing the stream name to ExampleOutputStream.

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

1. 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"))
   

2. Later in the tutorial, you run the stock.py script to send data to the application.

   $ python stock.py

1. Clone the remote repository using the following command:

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

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

To compile the application code

1. To use your application code, you compile and package it into a JAR file. You can compile and package your code in one of two ways:

   • Use the command-line Maven tool. Create your JAR file by running the following command in the directory that contains the pom.xml file:

     mvn package -Dflink.version=1.8.2

   • Use your development environment. See your development environment documentation for details.

     Note

     The provided source code relies on libraries from Java 1.8. Ensure that your project's Java version is 1.8.

   You can either upload your package as a JAR file, or you can compress your package and upload it as a ZIP file. If you create your application using the AWS CLI, you specify your code content type (JAR or ZIP).

2. If there are errors while compiling, verify that your JAVA_HOME environment variable is correctly set.

To upload the 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 the Configure options step, keep the settings as they are, and choose Next.

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

6. Choose Create bucket.

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

8. 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. Choose Next.

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Description, enter My java test app.

   • For Runtime, choose Apache Flink.

   • Leave the version pulldown as Apache Flink 1.8 (Recommended Version).

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

5. Choose Create application.

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": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/aws-kinesis-analytics-java-apps-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"
           }
       ]
   }

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. Enter the following application properties and values:

   Group ID	Key	Value
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

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

6. For CloudWatch logging, select the Enable check box.

7. Choose Update.

1. On the MyApplication page, choose Run. Confirm the action.

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

To create an IAM role

1. Open the IAM console at https://console.aws.amazon.com/iam/.

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

3. Under Select type of trusted identity, choose AWS Service. Under Choose the service that will use this role, choose Kinesis. Under Select your use case, choose Kinesis Analytics.

   Choose Next: Permissions.

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

5. On the Create role page, enter MF-stream-rw-role for the Role name. Choose Create role.

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

6. Attach the permissions policy to the role.

   Note

   For this exercise, Managed Service for Apache Flink 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 AKReadSourceStreamWriteSinkStream (the policy that you created in the previous section).

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

1. Save the following JSON code to a file named create_request.json. Replace the sample role ARN with the ARN for the role that you created previously. Replace the bucket ARN suffix (username) with the suffix that you chose in the previous section. Replace the sample account ID (012345678901) in the service execution role with your account ID.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_8",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. Execute the CreateApplication action with the preceding request to create the application:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

To start the application

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

   {
       "ApplicationName": "test",
       "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": "test"
   }
   

2. Execute the StopApplication action with the following 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": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

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 console at https://console.aws.amazon.com/kinesis.

2. In the Managed Service for Apache Flink panel, choose MyApplication.

3. Choose Configure.

4. In the Snapshots section, choose Disable and then choose Update.

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

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.aws.amazon.com/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.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-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. In the navigation bar, choose Roles.

7. Choose the kinesis-analytics-MyApplication-us-west-2 role.

8. Choose Delete role and then confirm the deletion.

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.

To sign up for an AWS account

1. Open https://portal.aws.amazon.com/billing/signup.

2. Follow the online instructions.

   Part of the sign-up procedure involves receiving a phone call and entering a verification code on the phone keypad.

   When you sign up for an AWS account, an AWS account root user is created. The root user has access to all AWS services and resources in the account. As a security best practice, assign administrative access to a user, and use only the root user to perform tasks that require root user access.

Secure your AWS account root user

1. Sign in to the AWS Management Console as the account owner by choosing Root user and entering your AWS account email address. On the next page, enter your password.

   For help signing in by using root user, see Signing in as the root user in the AWS Sign-In User Guide.

2. Turn on multi-factor authentication (MFA) for your root user.

   For instructions, see Enable a virtual MFA device for your AWS account root user (console) in the IAM User Guide.

Create a user with administrative access

1. Enable IAM Identity Center.

   For instructions, see Enabling AWS IAM Identity Center in the AWS IAM Identity Center User Guide.

2. In IAM Identity Center, grant administrative access to a user.

   For a tutorial about using the IAM Identity Center directory as your identity source, see Configure user access with the default IAM Identity Center directory in the AWS IAM Identity Center User Guide.

Sign in as the user with administrative access

• To sign in with your IAM Identity Center user, use the sign-in URL that was sent to your email address when you created the IAM Identity Center user.

  For help signing in using an IAM Identity Center user, see Signing in to the AWS access portal in the AWS Sign-In User Guide.

Assign access to additional users

1. In IAM Identity Center, create a permission set that follows the best practice of applying least-privilege permissions.

   For instructions, see Create a permission set in the AWS IAM Identity Center User Guide.

2. Assign users to a group, and then assign single sign-on access to the group.

   For instructions, see Add groups in the AWS IAM Identity Center User Guide.

To set up the AWS CLI

1. Download and configure the AWS CLI. For instructions, see the following topics in the AWS Command Line Interface User Guide:

   • Installing the AWS Command Line Interface

   • Configuring the AWS CLI

2. Add a named profile for the administrator user in the AWS CLI config file. You use this profile when executing the AWS CLI commands. For more information about named profiles, see Named Profiles in the AWS Command Line Interface User Guide.

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   For a list of available AWS Regions, see Regions and Endpoints in the Amazon Web Services General Reference.

   Note

   The example code and commands in this tutorial use the US West (Oregon) Region. To use a different Region, change the Region in the code and commands for this tutorial to the Region you want to use.

3. Verify the setup by entering the following help command at the command prompt:

   aws help

To create the data streams (AWS CLI)

1. To create the first stream (ExampleInputStream), use the following Amazon Kinesis create-stream AWS CLI command.

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

2. To create the second stream that the application uses to write output, run the same command, changing the stream name to ExampleOutputStream.

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-west-2 \
   --profile adminuser

1. 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"))
   

2. Later in the tutorial, you run the stock.py script to send data to the application.

   $ python stock.py

1. Clone the remote repository using the following command:

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

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

To compile the application code

1. To use your application code, you compile and package it into a JAR file. You can compile and package your code in one of two ways:

   • Use the command-line Maven tool. Create your JAR file by running the following command in the directory that contains the pom.xml file:

     mvn package

     Note

     The -Dflink.version parameter is not required for Managed Service for Apache Flink Runtime version 1.0.1; it is only required for version 1.1.0 and later. For more information, see Specify your application's Apache Flink version.

   • Use your development environment. See your development environment documentation for details.

   You can either upload your package as a JAR file, or you can compress your package and upload it as a ZIP file. If you create your application using the AWS CLI, you specify your code content type (JAR or ZIP).

2. If there are errors while compiling, verify that your JAVA_HOME environment variable is correctly set.

To upload the 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 the Configure options step, keep the settings as they are, and choose Next.

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

6. Choose Create bucket.

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

8. 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. Choose Next.

9. In the Set permissions step, keep the settings as they are. Choose Next.

10. In the Set properties step, keep the settings as they are. Choose Upload.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Description, enter My java test app.

   • For Runtime, choose Apache Flink.

     Note

     Managed Service for Apache Flink uses Apache Flink version 1.8.2 or 1.6.2.

   • Change the version pulldown to Apache Flink 1.6.

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

5. Choose Create application.

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": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/java-getting-started-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"
           }
       ]
   }

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 java-getting-started-1.0.jar.

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

4. Enter the following application properties and values:

   Group ID	Key	Value
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

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

6. For CloudWatch logging, select the Enable check box.

7. Choose Update.

1. On the MyApplication page, choose Run. Confirm the action.

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

To create an IAM role

1. Open the IAM console at https://console.aws.amazon.com/iam/.

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

3. Under Select type of trusted identity, choose AWS Service. Under Choose the service that will use this role, choose Kinesis. Under Select your use case, choose Kinesis Analytics.

   Choose Next: Permissions.

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

5. On the Create role page, enter MF-stream-rw-role for the Role name. Choose Create role.

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

6. Attach the permissions policy to the role.

   Note

   For this exercise, Managed Service for Apache Flink 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 AKReadSourceStreamWriteSinkStream (the policy that you created in the previous section).

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

1. Save the following JSON code to a file named create_request.json. Replace the sample role ARN with the ARN for the role that you created previously. Replace the bucket ARN suffix (username) with the suffix that you chose in the previous section. Replace the sample account ID (012345678901) in the service execution role with your account ID.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_6",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "java-getting-started-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. Execute the CreateApplication action with the preceding request to create the application:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

To start the application

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

   {
       "ApplicationName": "test",
       "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": "test"
   }
   

2. Execute the StopApplication action with the following 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": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

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 console at https://console.aws.amazon.com/kinesis.

2. In the Managed Service for Apache Flink panel, choose MyApplication.

3. Choose Configure.

4. In the Snapshots section, choose Disable and then choose Update.

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

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.aws.amazon.com/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.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-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. In the navigation bar, choose Roles.

7. Choose the kinesis-analytics-MyApplication-us-west-2 role.

8. Choose Delete role and then confirm the deletion.

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.

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

2. Run the stock.py script:

   $ python stock.py

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

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-examples.git

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

1. Install Java and Maven if you haven't already. For more information, see Complete the required prerequisites in the Tutorial: Get started using the DataStream API in Managed Service for Apache Flink tutorial.

2. Compile the application with the following command:

   mvn package -Dflink.version=1.15.3

   Note

   The provided source code relies on libraries from Java 11.

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.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Runtime, choose Apache Flink.

     Note

     Managed Service for Apache Flink uses Apache Flink version 1.15.2.

   • Leave the version pulldown 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.

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": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "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": "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"
           }
       ]
   }

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.

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.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. in the Managed Service for Apache Flink panel, choose MyApplication.

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

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.

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.

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-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. In the navigation bar, choose Roles.

7. Choose the kinesis-analytics-MyApplication-us-west-2 role.

8. Choose Delete role and then confirm the deletion.

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.

1. 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"))
   

2. Run the stock.py script:

   $ python stock.py

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

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-examples.git

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

1. Install Java and Maven if you haven't already. For more information, see Complete the required prerequisites in the Tutorial: Get started using the DataStream API in Managed Service for Apache Flink tutorial.

2. Compile the application with the following command:

   mvn package -Dflink.version=1.15.3

   Note

   The provided source code relies on libraries from Java 11.

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.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Runtime, choose Apache Flink.

   • Leave the version pulldown 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.

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": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "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": "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"
           }
       ]
   }

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.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. In the Managed Service for Apache Flink panel, choose MyApplication.

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

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.

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.

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-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. In the navigation bar, choose Roles.

7. Choose the kinesis-analytics-MyApplication-us-west-2 role.

8. Choose Delete role and then confirm the deletion.

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.

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

2. Run the stock.py script:

   $ python stock.py

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

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-examples.git

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

1. Install Java and Maven if you haven't already. For more information, see Complete the required prerequisites in the Tutorial: Get started using the DataStream API in Managed Service for Apache Flink tutorial.

2. Compile the application with the following command:

   mvn package -Dflink.version=1.15.3

1. In the Amazon S3 console, choose the ka-app-code-<username> bucket, navigate to the code folder, 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.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Runtime, choose Apache Flink.

   • Leave the version pulldown 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.

   Note

   When you create a Managed Service for Apache Flink 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:

   • For Application name, enter MyApplication.

   • For Runtime, choose Apache Flink.

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

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

7. Choose Create application.

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

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 code/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.

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.

1. 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;

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

1. 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.

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

   import java.util.concurrent.TimeUnit; 

3. 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.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. In the Managed Service for Apache Flink panel, choose MyApplication.

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

1. Open the Kinesis console at https://console.aws.amazon.com/kinesis.

2. In the Kinesis Data Streams panel, choose ExampleInputStream.

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

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.

1. Open the IAM console at https://console.aws.amazon.com/iam/.

2. On the navigation bar, choose Policies.

3. In the filter control, enter kinesis.

4. Choose the kinesis-analytics-service-MyApplication-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. On the navigation bar, choose Roles.

7. Choose the kinesis-analytics-MyApplication-us-west-2 role.

8. Choose Delete role and then confirm the deletion.

1. Open the CloudWatch console at https://console.aws.amazon.com/cloudwatch/.

2. On the navigation bar, choose Logs.

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

4. Choose Delete Log Group and then confirm the deletion.

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-examples.git

3. The application code is located in the amazon-kinesis-data-analytics-java-examples/KafkaConnectors/KafkaGettingStartedJob.java file. You can examine the code to familiarize yourself with the structure of Managed Service for Apache Flink application code.

4. Use either the command-line Maven tool or your preferred development environment to create the JAR file. To compile the JAR file using the command-line Maven tool, enter the following:

   mvn package -Dflink.version=1.15.3

   If the build is successful, the following file is created:

   target/KafkaGettingStartedJob-1.0.jar

   Note

   The provided source code relies on libraries from Java 11. If you are using a development environment,

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 KafkaGettingStartedJob-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.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink.

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Runtime, choose Apache Flink version 1.15.2.

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

5. Choose Create 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 KafkaGettingStartedJob-1.0.jar.

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

   Note

   When you specify application resources using the console (such as CloudWatch Logs or an Amazon VPC), the console modifies your application execution role to grant permission to access those resources.

4. Under Properties, choose Add Group. Enter the following properties:

   Group ID	Key	Value
   KafkaSource	topic	AWSKafkaTutorialTopic
   KafkaSource	bootstrap.servers	The bootstrap server list you saved previously
   KafkaSource	security.protocol	SSL
   KafkaSource	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   KafkaSource	ssl.truststore.password	changeit

   Note

   The ssl.truststore.password for the default certificate is "changeit"; you do not need to change this value if you are using the default certificate.

   Choose Add Group again. Enter the following properties:

   Group ID	Key	Value
   KafkaSink	topic	AWSKafkaTutorialTopicDestination
   KafkaSink	bootstrap.servers	The bootstrap server list you saved previously
   KafkaSink	security.protocol	SSL
   KafkaSink	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   KafkaSink	ssl.truststore.password	changeit
   KafkaSink	transaction.timeout.ms	1000

   The application code reads the above application properties to configure the source and sink used to interact with your VPC and Amazon MSK cluster. For more information about using properties, see Use runtime properties.

5. Under Snapshots, choose Disable. This will make it easier to update the application without loading invalid application state data.

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

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

8. In the Virtual Private Cloud (VPC) section, choose the VPC to associate with your application. Choose the subnets and security group associated with your VPC that you want the application to use to access VPC resources.

9. Choose Update.

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

2. Run the stock.py script:

   $ python stock.py

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

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-examples.git

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

1. Install Java and Maven if you haven't already. For more information, see Complete the required prerequisites in the Tutorial: Get started using the DataStream API in Managed Service for Apache Flink tutorial.

2. Compile the application with the following command:

   mvn package -Dflink.version=1.15.3

   Note

   The provided source code relies on libraries from Java 11.

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.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Runtime, choose Apache Flink.

     Note

     Managed Service for Apache Flink uses Apache Flink version 1.15.2.

   • Leave the version pulldown 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.

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.

   Note

   These permissions grant the application the ability to access the EFO consumer.

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "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": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "AllStreams",
               "Effect": "Allow",
               "Action": [
                   "kinesis:ListShards",
                   "kinesis:ListStreamConsumers",
                   "kinesis:DescribeStreamSummary"
               ],
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/*"
           },
           {
               "Sid": "Stream",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStream",
                   "kinesis:RegisterStreamConsumer",
                   "kinesis:DeregisterStreamConsumer"
               ],
               "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"
           },
           {
               "Sid": "Consumer",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:SubscribeToShard"
               ],
               "Resource": [
                   "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app",
                   "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app:*"
               ]
           }
       ]
   }

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 Properties, choose Create Group.

5. Enter the following application properties and values:

   Group ID	Key	Value
   ConsumerConfigProperties	flink.stream.recordpublisher	EFO
   ConsumerConfigProperties	flink.stream.efo.consumername	basic-efo-flink-app
   ConsumerConfigProperties	INPUT_STREAM	ExampleInputStream
   ConsumerConfigProperties	flink.inputstream.initpos	LATEST
   ConsumerConfigProperties	AWS_REGION	us-west-2

6. Under Properties, choose Create Group.

7. Enter the following application properties and values:

   Group ID	Key	Value
   ProducerConfigProperties	OUTPUT_STREAM	ExampleOutputStream
   ProducerConfigProperties	AWS_REGION	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

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

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

10. Choose Update.

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. in the Managed Service for Apache Flink panel, choose MyApplication.

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

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.

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.

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-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. In the navigation bar, choose Roles.

7. Choose the kinesis-analytics-MyApplication-us-west-2 role.

8. Choose Delete role and then confirm the deletion.

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.

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

2. Run the stock.py script:

   $ python stock.py

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

1. Clone the remote repository with the following command:

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

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

1. Install Java and Maven if you haven't already. For more information, see Complete the required prerequisites in the Tutorial: Get started using the DataStream API in Managed Service for Apache Flink tutorial.

2. In order to use the Kinesis connector for the following application, you need to download, build, and install Apache Maven. For more information, see Using the Apache Flink Kinesis Streams connector with previous Apache Flink versions.

3. Compile the application with the following command:

   mvn package -Dflink.version=1.15.3

   Note

   The provided source code relies on libraries from Java 11.

To upload the application code

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

2. In the console, choose the ka-app-code-<username> bucket, and then choose Upload.

3. In the Select files step, choose Add files. Navigate to the java-getting-started-1.0.jar file that you created in the previous step.

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. On the Managed Service for Apache Flink dashboard, choose Create analytics application.

3. On the Managed Service for Apache Flink - Create application page, provide the application details as follows:

   • For Application name, enter MyApplication.

   • For Description, enter My java test app.

   • For Runtime, choose Apache Flink.

     Note

     Managed Service for Apache Flink uses Apache Flink version 1.15.2.

   • Leave the version pulldown 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.

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 all the instances of 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/java-getting-started-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": "WriteDeliveryStream",
               "Effect": "Allow",
               "Action": "firehose:*",
               "Resource": "arn:aws:firehose:us-west-2:012345678901:deliverystream/ExampleDeliveryStream"
           }
       ]
   }

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 java-getting-started-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.

To create an IAM role

1. Open the IAM console at https://console.aws.amazon.com/iam/.

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

3. Under Select type of trusted identity, choose AWS Service. Under Choose the service that will use this role, choose Kinesis. Under Select your use case, choose Kinesis Analytics.

   Choose Next: Permissions.

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

5. On the Create role page, enter MF-stream-rw-role for the Role name. Choose Create role.

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

6. Attach the permissions policy to the role.

   Note

   For this exercise, Managed Service for Apache Flink 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 AKReadSourceStreamWriteSinkStream (the policy that you created in the previous section).

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

1. Save the following JSON code to a file named create_request.json. Replace the sample role ARN with the ARN for the role that you created previously. Replace the bucket ARN suffix with the suffix that you chose in the Create dependent resources section (ka-app-code-<username>.) Replace the sample account ID (012345678901) in the service execution role with your account ID.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_15",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "java-getting-started-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           }
         }
       }
   }
   
   

2. Execute the CreateApplication action with the preceding request to create the application:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

To start the application

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

   {
       "ApplicationName": "test",
       "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": "test"
   }
   

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

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

1. Open the Managed Service for Apache Flink console at https://console.aws.amazon.com/flink

2. In the Managed Service for Apache Flink panel, choose MyApplication.

3. Choose Configure.

4. In the Snapshots section, choose Disable and then choose Update.

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

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.

1. Open the Kinesis console at https://console.aws.amazon.com/kinesis.

2. In the Firehose panel, choose ExampleDeliveryStream.

3. In the ExampleDeliveryStream page, choose Delete Firehose stream and then confirm the deletion.

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.

4. If you created an Amazon S3 bucket for your Firehose stream's destination, delete that bucket too.

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-us-west-2 policy.

5. Choose Policy Actions and then choose Delete.

6. If you created a new policy for your Firehose stream, delete that policy too.

7. In the navigation bar, choose Roles.

8. Choose the kinesis-analytics-MyApplication-us-west-2 role.

9. Choose Delete role and then confirm the deletion.

10. If you created a new role for your Firehose stream, delete that role too.

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.