Create and Run a Kinesis Data Analytics for Python 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. 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-java-examples

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

To upload the application code using the console:

1. Use your preferred compression application to compress the getting-started.py and https://mvnrepository.com/artifact/org.apache.flink/flink- sql-connector-kinesis_2.12/1.15.2 files. Name the archive myapp.zip. If you include the outer folder in your archive, you must include this in the path with the code in your configuration file(s): GettingStarted/getting-started.py.

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

3. Choose Create bucket.

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

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

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

7. Choose Create bucket.

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

9. In the Select files step, choose Add files. Navigate to the myapp.zip file that you created in the previous step. Choose Next.

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

To upload the application code using the Amazon CLI:

1. Use your preferred compression application to compress the streaming-file-sink.py and https://mvnrepository.com/artifact/org.apache.flink/flink- sql-connector-kinesis_2.12/1.15.2 files.

   Note

   Do not use the compress features in Finder (macOS) or Windows Explorer (Windows) to create the myapp.zip archive. This may result in invalid application code.

2. Use your preferred compression application to compress the getting-started.py and https://mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-kinesis/1.15.2 files. Name the archive myapp.zip. If you include the outer folder in your archive, you must include this in the path with the code in your configuration file(s): GettingStarted/getting-started.py.

3. Run the following command:

   $ aws s3 --region aws region cp myapp.zip s3://ka-app-code-<username>

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

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

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

   • For Application name, enter MyApplication.

   • For Description, enter My java test app.

   • For Runtime, choose Apache Flink.

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

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

5. Choose Create application.

To configure the application

1. On the MyApplication page, choose Configure.

2. On the Configure application page, provide the Code location:

   • For Amazon S3 bucket, enter ka-app-code-<username>.

   • For Path to Amazon S3 object, enter myapp.zip.

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

4. Under Properties, choose Add group.

5. Enter the following:

   Group ID	Key	Value
   consumer.config.0	input.stream.name	ExampleInputStream
   consumer.config.0	aws.region	us-west-2
   consumer.config.0	scan.stream.initpos	LATEST

   Choose Save.

6. Under Properties, choose Add group again.

7. Enter the following:

   Group ID	Key	Value
   producer.config.0	output.stream.name	ExampleOutputStream
   producer.config.0	aws.region	us-west-2
   producer.config.0	shard.count	1

8. Under Properties, choose Add group again. For Group ID, enter kinesis.analytics.flink.run.options. This special property group tells your application where to find its code resources. For more information, see Specifying your Code Files.

9. Enter the following:

   Group ID	Key	Value
   kinesis.analytics.flink.run.options	python	getting-started.py
   kinesis.analytics.flink.run.options	jarfile	flink-sql-connector-kinesis-1.15.2.jar

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

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

12. Choose Update.

To edit the IAM policy to add S3 bucket permissions

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

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

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

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

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-username/myapp.zip"
               ]
           },
           {
               "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"
           }
       ]
   }