<jupyter_start><jupyter_text>Real Time Processing - Part 2 Using a DynamoDB table to capture specific tweet aspects from Kinesis  This notebook provides the necessary code to create a dynamoDB table, with the hashtag as the partition key (Primary Key). Creating the DynamoDB table, with the hashtag as the primary key. Read and Write capacity provisioning is kept low. The boto3 library is used to interface to Jupyter<jupyter_code>## create a table to store twitter hashtags in DynamoDB
import boto3
dynamodb = boto3.resource('dynamodb')
table = dynamodb.create_table(
    TableName='hashtags',
    KeySchema=[
        {
            'AttributeName': 'hashtag',
            'KeyType': 'HASH'
        }
    ],
    AttributeDefinitions=[
        {
            'AttributeName': 'hashtag',
            'AttributeType': 'S'
        }
    ],
    # the pricing isdetermined by Provisioned Throughput, thus it is kept low
    ProvisionedThroughput={
        'ReadCapacityUnits': 100,
        'WriteCapacityUnits': 100
    }
)
table.meta.client.get_waiter('table_exists').wait(TableName='hashtags')<jupyter_output><empty_output><jupyter_text>Linking the Kinesis stream to DynamoDB, parsing and storing tweets Parsing the results from Kinesis in JSON , extracting the hashtag and submitting the input into table.update_item(…). The table.update_item is a larger request in JSON to increment by 1 if it exists, and put 1 otherwise (not increment).Thus the hashtag counter increases by 1 each time it detects the hashtag associated with it (for every tweet)<jupyter_code>## Importing the necessary libraries

import time
import json
import decimal

#Connecting to the kinesis stream-need to specify kinesis stream here
kinesis = boto3.client("kinesis")
shard_id = 'shardId-000000000000' #only one shard
shard_it = kinesis.get_shard_iterator(StreamName="twitter_bigdata", ShardId=shard_id, ShardIteratorType="TRIM_HORIZON")["ShardIterator"]

#connecting to the dynamoDB table
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('hashtags')
#Parsing the tweets and setting a counter

while 1==1:
    out = kinesis.get_records(ShardIterator=shard_it, Limit=100)
    for record in out['Records']:
        if 'entities' in json.loads(record['Data']):
            htags = json.loads(record['Data'])['entities']['hashtags']
            if htags:
                    for ht in htags:
                        htag = ht['text']
                        checkItemExists = table.get_item(
                           Key={
                                        'hashtag':htag
                                }
                            )
                    if 'Item' in checkItemExists:
                        response = table.update_item(
                                Key={
                                    'hashtag': htag 
                            }, #updating the counter if hashtags exists
                            UpdateExpression="set htCount  = htCount + :val",
                            ConditionExpression="attribute_exists(hashtag)",
                            ExpressionAttributeValues={
                                ':val': decimal.Decimal(1) 	
                            },
                            ReturnValues="UPDATED_NEW"
                        )
                    else: #not updating the counter if hashtags not exist
                                        response = table.update_item(
                                                Key={
                                                        'hashtag': htag
                                                },
                                                UpdateExpression="set htCount = :val",
                                                ExpressionAttributeValues={
                                                        ':val': decimal.Decimal(1)
                                                },
                                                ReturnValues="UPDATED_NEW"
                                        )    
    shard_it = out["NextShardIterator"]
    time.sleep(1.0)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07383040935672515, 'fraction_numerical': 0.008771929824561403, 'mean_word_length': 1.5997466751108296, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '390472', 'license': 'Apache-2.0', 'n_tokens_mistral': 1044, 'n_tokens_neox': 973, 'n_words': 293, 'path': 'DynamoDB-Twitter-Realtimestreaming-Deletion.ipynb', 'repo_name': 'wowlenin/sp20-516-253', 'size': '4104'}