<jupyter_start><jupyter_text>Kinesis Data Analytics![](img/kinesis_data_analytics_docs.png)<jupyter_code>import boto3
import sagemaker
import pandas as pd

sess   = sagemaker.Session()
bucket = sess.default_bucket()
role = sagemaker.get_execution_role()
region = boto3.Session().region_name

kinesis = boto3.Session().client(service_name='kinesis', region_name=region)
analytics = boto3.client('kinesisanalyticsv2')
sts = boto3.Session().client(service_name='sts', region_name=region)<jupyter_output><empty_output><jupyter_text>Download Dataset<jupyter_code>!aws s3 cp 's3://amazon-reviews-pds/tsv/amazon_reviews_us_Digital_Software_v1_00.tsv.gz' ./data/
import csv
import pandas as pd

df = pd.read_csv('./data/amazon_reviews_us_Digital_Software_v1_00.tsv.gz', 
                 delimiter='\t', 
                 quoting=csv.QUOTE_NONE,
                 compression='gzip')
df.shape
df.head(5)
partition_key = '123'
#reviews_tsv = '5\tThis is a 5 star review\n1\tThis is a 1 star review\n'
df_star_rating_and_review_body = df[['star_rating', 'review_body']][:100]
df_star_rating_and_review_body.shape
df_star_rating_and_review_body.head()
reviews_tsv = df_star_rating_and_review_body.to_csv(sep='\t',
                                                    header=None,
                                                    index=False)
reviews_tsv
%store -r iam_role_kinesis_arn
print(iam_role_kinesis_arn)
%store -r data_stream_arn
print(data_stream_arn)
%store -r firehouse_delivery_stream_arn
print(firehouse_delivery_stream_arn)<jupyter_output>arn:aws:firehose:ap-southeast-2:979770387749:deliverystream/dsoaws-firehose-stream<jupyter_text>TODO:  Data Analyticshttps://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kinesisanalyticsv2.htmlKinesisAnalyticsV2.Client.create_application<jupyter_code>response = analytics.create_application(
    ApplicationName='DSOAWS_Kinesis_Data_Analytics_Application',
#    ApplicationDescription='string',
    RuntimeEnvironment='SQL-1_0',
#    RuntimeEnvironment='SQL-1_0'|'FLINK-1_6'|'FLINK-1_8',
    ServiceExecutionRole=iam_role_kinesis_arn,
    ApplicationConfiguration={
        'SqlApplicationConfiguration': {
            'Inputs': [
                {
                    'NamePrefix': 'DSOAWS_Kinesis_Data_Analytics_Application_Output',
                    'InputProcessingConfiguration': {
                        'InputLambdaProcessor': {
                            'ResourceARN': 'string'
                        }
                    },
                    'KinesisStreamsInput': {
                        'ResourceARN': data_stream_arn
                    },
#                    'KinesisFirehoseInput': {
#                        'ResourceARN': 'string'
#                    },
                    'InputParallelism': {
                        'Count': 1
                    },
                    'InputSchema': {
                        'RecordFormat': {
                            'RecordFormatType': 'CSV',
                            'MappingParameters': {
#                                 'JSONMappingParameters': {
#                                     'RecordRowPath': 'string'
#                                 },
                                'CSVMappingParameters': {
                                    'RecordRowDelimiter': '\n',
                                    'RecordColumnDelimiter': '\t'
                                }
                            }
                        },
                        'RecordEncoding': 'UTF-8',
                        'RecordColumns': [
                            {
                                'Name': 'star_rating',
                                'SqlType': 'VARCHAR'
                            },
                            {
                                'Name': 'review_body',
                                'SqlType': 'VARCHAR'
                            },                            
                        ]
                    }
                },
            ],
            'Outputs': [
                {
                    'Name': 'DSOAWS_Kinesis_Data_Analytics_Application_Output',
#                     'KinesisStreamsOutput': {
#                         'ResourceARN': 'string'
#                     },
                    'KinesisFirehoseOutput': {
                        'ResourceARN': firehouse_delivery_stream_arn
                    },
#                     'LambdaOutput': {
#                         'ResourceARN': 'string'
#                     },
                    'DestinationSchema': {
                        'RecordFormatType': 'CSV'
                    }
                },
            ],
#             'ReferenceDataSources': [
#                 {
#                     'TableName': 'string',
#                     'S3ReferenceDataSource': {
#                         'BucketARN': 'string',
#                         'FileKey': 'string'
#                     },
#                     'ReferenceSchema': {
#                         'RecordFormat': {
#                             'RecordFormatType': 'JSON'|'CSV',
#                             'MappingParameters': {
#                                 'JSONMappingParameters': {
#                                     'RecordRowPath': 'string'
#                                 },
#                                 'CSVMappingParameters': {
#                                     'RecordRowDelimiter': 'string',
#                                     'RecordColumnDelimiter': 'string'
#                                 }
#                             }
#                         },
#                         'RecordEncoding': 'string',
#                         'RecordColumns': [
#                             {
#                                 'Name': 'string',
#                                 'Mapping': 'string',
#                                 'SqlType': 'string'
#                             },
#                         ]
#                     }
#                 },
#             ]            
        },
        'ApplicationCodeConfiguration' : {
            'CodeContentType': 'PLAINTEXT'
        }
    }
)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09959514170040486, 'fraction_numerical': 0.008582995951417004, 'mean_word_length': 1.1459346768589298, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 12, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 19, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '256027', 'license': 'Apache-2.0', 'n_tokens_mistral': 1632, 'n_tokens_neox': 1526, 'n_words': 179, 'path': '11_stream/archive/05_Kinesis_Data_Analytics.ipynb', 'repo_name': 'MarcusFra/workshop', 'size': '6175'}