Streaming Ingestion with Kinesis

Introduction: Moving from Batch to Real-Time Ingestion

Welcome back! In the previous lesson, you built a strong foundation for your AWS data lake by designing a well-organized structure on Amazon S3 . You learned how to create a multi-zone layout with raw, processed, and curated data, and you uploaded sample data using time-based partitioning. This setup is perfect for batch analytics, where data is collected and processed in chunks. However, many modern applications need to process data as soon as it arrives. For example, you might want to analyze user activity on your website in real time, detect fraud as transactions happen, or monitor sensor data from IoT devices without delay. This is where real-time data ingestion becomes important. To enable real-time data flows into your data lake, AWS offers a service called Amazon Kinesis Data Streams . Kinesis allows you to collect, process, and move data continuously, so you can react to new information instantly. In this lesson, you will learn how to create a Kinesis Data Stream using Python . This is the first step in bridging the gap between real-time data sources and your S3-based data lake.

Kinesis Data Streams: Key Concepts

Before you start working with Kinesis, let’s clarify a few important concepts. A Kinesis Data Stream is a managed service that lets you collect and transport data records in real time. Think of it as a pipeline that carries small pieces of data — called records — from producers (like your application or website) to consumers (such as analytics tools or storage systems). Each stream is made up of one or more shards . A shard is like a lane on a highway: it determines how much data can flow through the stream at once. If you need to handle more data, you can add more shards, just like adding more lanes to a road allows more cars to travel at the same time. For most beginner projects, starting with a single shard is enough. When you create a Kinesis stream, it goes through different status phases. At first, the stream is in the CREATING state while AWS sets up the necessary resources. Once it is ready, the status changes to ACTIVE, and you can start sending and receiving data. It is important to wait for the stream to become ACTIVE before using it; otherwise, your application may run into errors. Here is a quick summary of the key terms: Stream: The main resource that collects and transports your data records. Record: A single piece of data, such as a JSON object or a string. Shard: A unit of capacity within the stream. More shards mean more data can be handled at once. Status: The current state of the stream, such as CREATING or ACTIVE . Understanding these concepts will help you as you start working with Kinesis and connect it to your data lake.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Creating a Kinesis Data Stream with Python

Let’s walk through how to create a Kinesis Data Stream using Python and the boto3 library. This example will show you how to set up a stream called user-events-stream with one shard, wait for it to become active, and handle common errors, such as trying to create a stream that already exists.

On CodeSignal, the boto3 library is already installed, so you do not need to worry about installation here. If you are working on your own device, you can install it using pip install boto3.

Here is the complete code example:

Python

import boto3
import uuid
import os
import time
from botocore.exceptions import ClientError

# Configuration
with open(".codesignal/suffix.txt") as f:
    SUFFIX = f.read().strip()
STREAM_NAME = f'user-events-stream-{SUFFIX}'  # Name of the Kinesis stream to create
SHARD_COUNT = 1  # Number of shards (capacity units) for the stream

def create_kinesis_stream():
    """Create Kinesis Data Stream"""
    # Create a Kinesis client using boto3
    kinesis_client = boto3.client('kinesis')
    
    try:
        # Attempt to create the stream with the specified name and shard count
        response = kinesis_client.create_stream(
            StreamName=STREAM_NAME,
            ShardCount=SHARD_COUNT
        )
        
        print(f"Creating stream: {STREAM_NAME}")
        print("Waiting for stream to become active...")
        
        # Use a waiter to pause execution until the stream exists
        waiter = kinesis_client.get_waiter('stream_exists')
        waiter.wait(StreamName=STREAM_NAME)
        
        # Retrieve information about the stream to check its status
        stream_info = kinesis_client.describe_stream(StreamName=STREAM_NAME)
        status = stream_info['StreamDescription']['StreamStatus']
        
        print(f"Stream status: {status}")
        
        # If the stream is active, print its details
        if status == 'ACTIVE':
            print("Stream is ready for data!")
            print(f"Stream ARN: {stream_info['StreamDescription']['StreamARN']}")
            print(f"Shard count: {len(stream_info['StreamDescription']['Shards'])}")
        
    except ClientError as e:
        code = e.response["Error"]["Code"]
        # Handle the case where the stream already exists
        if code == 'ResourceInUseException':
            print(f"Stream {STREAM_NAME} already exists")
        # Handle any other client errors that may occur
        else:
            message = e.response["Error"].get("Message", str(e))
            print(f"Error creating stream: {code}: {message}")

if __name__ == "__main__":
    # Run the function to create the Kinesis stream
    create_kinesis_stream()

import boto3
import uuid
import os
import time
from botocore.exceptions import ClientError

# Configuration
with open(".codesignal/suffix.txt") as f:
    SUFFIX = f.read().strip()
STREAM_NAME = f'user-events-stream-{SUFFIX}'  # Name of the Kinesis stream to create
SHARD_COUNT = 1  # Number of shards (capacity units) for the stream

def create_kinesis_stream():
    """Create Kinesis Data Stream"""
    # Create a Kinesis client using boto3
    kinesis_client = boto3.client('kinesis')
    
    try:
        # Attempt to create the stream with the specified name and shard count
        response = kinesis_client.create_stream(
            StreamName=STREAM_NAME,
            ShardCount=SHARD_COUNT
        )
        
        print(f"Creating stream: {STREAM_NAME}")
        print("Waiting for stream to become active...")
        
        # Use a waiter to pause execution until the stream exists
        waiter = kinesis_client.get_waiter('stream_exists')
        waiter.wait(StreamName=STREAM_NAME)
        
        # Retrieve information about the stream to check its status
        stream_info = kinesis_client.describe_stream(StreamName=STREAM_NAME)
        status = stream_info['StreamDescription']['StreamStatus']
        
        print(f"Stream status: {status}")
        
        # If the stream is active, print its details
        if status == 'ACTIVE':
            print("Stream is ready for data!")
            print(f"Stream ARN: {stream_info['StreamDescription']['StreamARN']}")
            print(f"Shard count: {len(stream_info['StreamDescription']['Shards'])}")
        
    except ClientError as e:
        code = e.response["Error"]["Code"]
        # Handle the case where the stream already exists
        if code == 'ResourceInUseException':
            print(f"Stream {STREAM_NAME} already exists")
        # Handle any other client errors that may occur
        else:
            message = e.response["Error"].get("Message", str(e))
            print(f"Error creating stream: {code}: {message}")

if __name__ == "__main__":
    # Run the function to create the Kinesis stream
    create_kinesis_stream()

Let’s break down what this code does. First, it creates a Kinesis client using boto3, which allows you to interact with AWS services from Python. The code then tries to create a new stream with the specified name and shard count. After sending the creation request, it uses a special tool called a waiter to pause the program until the stream is fully set up and ready to use. Once the stream is active, the code fetches and prints details about the stream, such as its status, Amazon Resource Name (ARN), and the number of shards. If the stream already exists, the code handles this gracefully and prints a helpful message.

When you run this code, you should see output similar to the following if the stream is created successfully:

text

Creating stream: user-events-stream
Waiting for stream to become active...
Stream status: ACTIVE
Stream is ready for data!
Stream ARN: arn:aws:kinesis:us-east-1:123456789012:stream/user-events-stream
Shard count: 1

Creating stream: user-events-stream
Waiting for stream to become active...
Stream status: ACTIVE
Stream is ready for data!
Stream ARN: arn:aws:kinesis:us-east-1:123456789012:stream/user-events-stream
Shard count: 1

If the stream already exists, you will see:

text

Stream user-events-stream already exists

Stream user-events-stream already exists

This example gives you a solid starting point for working with Kinesis streams in your own projects.

Understanding the Partition Key in Kinesis

When you send a record to a Kinesis Data Stream, you must specify a partition key. The partition key is a string that Kinesis uses to determine which shard in the stream will store the record. This is a crucial concept for both performance and data organization.

How the Partition Key Works:
Kinesis uses the partition key to compute a hash value, which maps the record to a specific shard in the stream. All records with the same partition key will always go to the same shard.
Why the Partition Key Matters:
- Load Distribution: If you use the same partition key for all records, all your data will go to a single shard, which can quickly become a bottleneck. To take advantage of multiple shards (and thus higher throughput), you should use a partition key that distributes records evenly across shards.
- Ordering Guarantees: Kinesis guarantees that records with the same partition key are stored in order within a shard. This is important if you need to process events for a particular user or device in the exact order they occurred.
- Scalability: Choosing a good partition key helps you scale your stream. For example, using a user ID, device ID, or order ID as the partition key can help distribute records more evenly, especially if you have many unique users or devices.
Example: Choosing a Partition Key:
In the order event example below, we use the user_id as the partition key. If you have many users placing orders, this will distribute the records across shards based on the user ID. If you used a constant value (like "orders"), all records would go to the same shard, which is not ideal for high-throughput scenarios.
Best Practices:
- High Cardinality: Choose a partition key with high cardinality (many unique values) to maximize parallelism.
- Consistent Usage: Use the same logic for partition keys across your producers to ensure predictable data distribution.
- Ordering Needs: If you need to process all events for a specific entity in order, use that entity’s ID as the partition key.

Understanding and choosing the right partition key is essential for building scalable, efficient, and reliable streaming data pipelines with Kinesis.

Sending Data Records to the Kinesis Stream

Now that you have created a Kinesis Data Stream, let’s see how to send data into it. In Kinesis terminology, this is called putting records into the stream. Each record can represent an event, such as a user action or a sensor reading.

Here’s a simple example that demonstrates how to send a single order event (as a JSON object) to the user-events-stream using Python and boto3:

Python

import boto3
import uuid
import os
import json
import time

# Configuration
with open(".codesignal/suffix.txt") as f:
    SUFFIX = f.read().strip()
STREAM_NAME = f'user-events-stream-{SUFFIX}'

def put_order_event(order_id, user_id, amount):
    """Send a single order event to the Kinesis stream."""
    kinesis_client = boto3.client('kinesis')
    
    # Create a sample order event as a Python dictionary
    order_event = {
        "order_id": order_id,
        "user_id": user_id,
        "amount": amount,
        "timestamp": int(time.time())
    }
    
    # Convert the event to a JSON string and then to bytes
    data = json.dumps(order_event).encode('utf-8')
    
    # Use user_id as the partition key to distribute records across shards
    response = kinesis_client.put_record(
        StreamName=STREAM_NAME,
        Data=data,
        PartitionKey=str(user_id)
    )
    
    print(f"Put order event: {order_event}")
    print(f"Record ID: {response['SequenceNumber']}")

if __name__ == "__main__":
    # Example: send a sample order event
    put_order_event(order_id="ORD123", user_id="U456", amount=99.99)

import boto3
import uuid
import os
import json
import time

# Configuration
with open(".codesignal/suffix.txt") as f:
    SUFFIX = f.read().strip()
STREAM_NAME = f'user-events-stream-{SUFFIX}'

def put_order_event(order_id, user_id, amount):
    """Send a single order event to the Kinesis stream."""
    kinesis_client = boto3.client('kinesis')
    
    # Create a sample order event as a Python dictionary
    order_event = {
        "order_id": order_id,
        "user_id": user_id,
        "amount": amount,
        "timestamp": int(time.time())
    }
    
    # Convert the event to a JSON string and then to bytes
    data = json.dumps(order_event).encode('utf-8')
    
    # Use user_id as the partition key to distribute records across shards
    response = kinesis_client.put_record(
        StreamName=STREAM_NAME,
        Data=data,
        PartitionKey=str(user_id)
    )
    
    print(f"Put order event: {order_event}")
    print(f"Record ID: {response['SequenceNumber']}")

if __name__ == "__main__":
    # Example: send a sample order event
    put_order_event(order_id="ORD123", user_id="U456", amount=99.99)

What this code does:

Defines a function put_order_event that takes order details and sends them as a record to the Kinesis stream.
Serializes the order event as JSON and encodes it as bytes (required by Kinesis).
Uses the put_record method to send the data, specifying a PartitionKey (here, the user_id), which determines how records are distributed across shards.
Prints confirmation with the event data and the record’s sequence number.

Sample output:

text

Put order event: {'order_id': 'ORD123', 'user_id': 'U456', 'amount': 99.99, 'timestamp': 1717690000}
Record ID: 49590338271490256608559692538361571095921575989136588898

Put order event: {'order_id': 'ORD123', 'user_id': 'U456', 'amount': 99.99, 'timestamp': 1717690000}
Record ID: 49590338271490256608559692538361571095921575989136588898

You can call put_order_event multiple times (with different order details) to simulate a stream of real-time order events flowing into your data lake pipeline. This is the basic building block for ingesting real-time data into your AWS data lake using Kinesis.

Introduction: Moving from Batch to Real-Time Ingestion

Kinesis Data Streams: Key Concepts

Join the 1M+ learners on CodeSignal

Shard limits and cost

Creating a Kinesis Data Stream with Python

Understanding the Partition Key in Kinesis

Sending Data Records to the Kinesis Stream

How Kinesis Bridges to Your Data Lake

Summary and What’s Next