<jupyter_start><jupyter_text>Amazon Kinesis Data StreamAmazon Kinesis Data Streams ingests a large amount of data in real time, durably stores the data, and makes the data available for consumption. The unit of data stored by Kinesis Data Streams is a data record. A data stream represents a group of data records. The data records in a data stream are distributed into shards.A shard has a sequence of data records in a stream. When you create a stream, you specify the number of shards for the stream. The total capacity of a stream is the sum of the capacities of its shards. You can increase or decrease the number of shards in a stream as needed. However, you are charged on a per-shard basis. The producers continually push data to Kinesis Data Streams, and the consumers process the data in real time. Consumers (such as a custom application running on Amazon EC2 or an Amazon Kinesis Data Firehose delivery stream) can store their results using an AWS service such as Amazon DynamoDB, Amazon Redshift, or Amazon S3.<jupyter_code>import boto3
import sagemaker
import pandas as pd
import json

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

sm = boto3.Session().client(service_name="sagemaker", region_name=region)
kinesis = boto3.Session().client(service_name="kinesis", region_name=region)
sts = boto3.Session().client(service_name="sts", region_name=region)<jupyter_output><empty_output><jupyter_text>Create a Kinesis Data Stream<jupyter_code>%store -r stream_name
try:
    stream_name
except NameError:
    print("+++++++++++++++++++++++++++++++")
    print("[ERROR] Please run all previous notebooks in this section before you continue.")
    print("+++++++++++++++++++++++++++++++")
print(stream_name)
shard_count = 2
from botocore.exceptions import ClientError

try:
    response = kinesis.create_stream(StreamName=stream_name, ShardCount=shard_count)
    print("Data Stream {} successfully created.".format(stream_name))
    print(json.dumps(response, indent=4, sort_keys=True, default=str))

except ClientError as e:
    if e.response["Error"]["Code"] == "ResourceInUseException":
        print("Data Stream {} already exists.".format(stream_name))
    else:
        print("Unexpected error: %s" % e)
import time

status = ""
while status != "ACTIVE":
    r = kinesis.describe_stream(StreamName=stream_name)
    description = r.get("StreamDescription")
    status = description.get("StreamStatus")
    time.sleep(5)

print("Stream {} is active".format(stream_name))<jupyter_output><empty_output><jupyter_text>_This may take a minute.  Please be patient._<jupyter_code>stream_response = kinesis.describe_stream(StreamName=stream_name)

print(json.dumps(stream_response, indent=4, sort_keys=True, default=str))
stream_arn = stream_response["StreamDescription"]["StreamARN"]
print(stream_arn)
%store stream_arn<jupyter_output><empty_output><jupyter_text>Review Kinesis Data Stream<jupyter_code>from IPython.core.display import display, HTML

display(
    HTML(
        '<b>Review <a target="blank" href="https://console.aws.amazon.com/kinesis/home?region={}#/streams/details/{}/details"> Kinesis Data Stream</a></b>'.format(
            region, stream_name
        )
    )
)<jupyter_output><empty_output><jupyter_text>Store Variables for the Next Notebooks<jupyter_code>%store
%%javascript
Jupyter.notebook.save_checkpoint()
Jupyter.notebook.session.delete();<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11209862385321101, 'fraction_numerical': 0.0031536697247706424, 'mean_word_length': 5.722543352601156, 'pattern_counts': {'":': 2, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '439064', 'license': 'Apache-2.0', 'n_tokens_mistral': 1054, 'n_tokens_neox': 1037, 'n_words': 319, 'path': '11_stream/archive/11_stream.orig/03_Create_Kinesis_Data_Stream.ipynb', 'repo_name': 'NRauschmayr/workshop', 'size': '3488'}