Step 3: Create and run a Managed Service for Apache Flink application

To create the data streams (Amazon CLI)

1. To create the first stream (ExampleInputStream), use the following Amazon Kinesis create-stream Amazon 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-managed-service-for-apache-flink-examples.git

2. Navigate to the amazon-managed-service-for-apache-flink-examples/tree/main/java/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.19.1

   • 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 Amazon 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.amazonaws.cn/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 amazon-msf-java-stream-app-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.19.

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.amazonaws.cn/iam/.

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

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

4. Add the highlighted section of the following policy example to the policy. Replace the sample account IDs (012345678901) with your account ID.

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/amazon-msf-java-stream-app-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 amazon-msf-java-stream-app-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
   FlinkApplicationProperties	flink.inputstream.initpos	LATEST
   FlinkApplicationProperties	aws.region	us-west-2
   FlinkApplicationProperties	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.amazonaws.cn/iam/.

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

3. Under Select type of trusted identity, choose Amazon 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_19",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "amazon-msf-java-stream-app-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