Kinesis vs SQS: Order Processing

Build an order pipeline using Kinesis streams and SQS queues together.

Introduction

30 Second Summary

Think about an online store processing hundreds of orders per second. Some tasks need to happen in a specific sequence (charge the card before shipping), while others can happen independently (sending a confirmation email does not depend on updating inventory).

In this project, you will build a mini e-commerce order processing pipeline using AWS Kinesis Data Streams and SQS together. You will see firsthand why real-world systems use both services, not one or the other.

What You'll Build

A three-component pipeline that ingests real-time order events into a Kinesis stream, processes them in order, and fans out independent tasks to an SQS queue for parallel handling.

By the end of this project, you'll have:

  • A working Kinesis Data Stream that preserves the order of events using partition keys.
  • A producer and processor pair that proves Kinesis ordering guarantees in action.
  • An SQS queue with a worker that handles decoupled tasks independently.
  • Secret Mission: add a Dead Letter Queue that catches failed tasks and implement retry logic for production-grade error handling.

Are there any prerequisites?

You will need an existing AWS account and a basic coding background in Python. No prior experience with Kinesis or SQS is needed.

Set Up Your Environment

Before you can build anything with AWS services, your local machine needs the tools to talk to the cloud. The AWS CLI is your command-line connection to AWS, and boto3 is the Python library that lets your scripts interact with services like Kinesis and SQS programmatically.

In this step, you will install and configure all three tools so your scripts can create and manage AWS resources from your terminal.

In this step, get ready to:
  • Install and verify AWS CLI v2.
  • Verify Python 3 and install the boto3 library.
  • Configure AWS CLI credentials for your IAM user.
Install and verify AWS CLI v2

The AWS CLI is the command-line tool that lets you manage AWS services directly from your terminal. You need version 2 because it bundles all dependencies and supports the latest AWS features.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), then typing Terminal (macOS) or cmd (Windows) and pressing Enter.
  • Check if AWS CLI v2 is already installed by running this command:
aws --version

Look at the output in your terminal and choose the tab that matches what you see.

✔️ I see aws-cli/2.x.x

Your AWS CLI v2 is already installed and ready to go. Move on to the next substep.

ⓧ I see aws-cli/1.x.x

You have AWS CLI version 1 installed. Version 2 is a separate install that will replace it. Download and install the v2 package for your platform.

macOS

  • Download the official installer by running:
curl "https://awscli.amazonaws.com/AWSCLIV2.pkg" -o "AWSCLIV2.pkg"
  • Run the installer:
sudo installer -pkg AWSCLIV2.pkg -target /

The installer will ask for your system password because it installs to a system directory.

  • Verify the upgrade by running:
aws --version

You should now see aws-cli/2.x.x in the output.

Windows

  • Download the MSI installer from:
https://awscli.amazonaws.com/AWSCLIV2.msi
  • Double-click the downloaded file and follow the on-screen prompts to complete the installation.
  • Close and reopen your command prompt, then verify by running:
aws --version

You should now see aws-cli/2.x.x in the output.

ⓧ Command not found

AWS CLI is not installed yet. Install it for your platform.

macOS

  • Download the official installer by running:
curl "https://awscli.amazonaws.com/AWSCLIV2.pkg" -o "AWSCLIV2.pkg"
  • Run the installer:
sudo installer -pkg AWSCLIV2.pkg -target /

The installer will ask for your system password because it installs to a system directory.

  • Verify the installation by running:
aws --version

You should see aws-cli/2.x.x in the output.

Windows

  • Download the MSI installer from:
https://awscli.amazonaws.com/AWSCLIV2.msi
  • Double-click the downloaded file and follow the on-screen prompts to complete the installation.
  • Close and reopen your command prompt, then verify by running:
aws --version

You should see aws-cli/2.x.x in the output.

Still not seeing a version number?

If you still get "command not found" after installing, close your terminal completely and open a new one. The install updates your system PATH, but existing terminal sessions do not pick it up automatically.

help me troubleshoot AWS CLI installation

Verify Python 3 and install boto3

Your scripts need Python 3 to run. The boto3 library is the official AWS SDK for Python that handles all the API calls to Kinesis and SQS.

  • Check that Python 3 is available by running:
python3 --version

✔️ I see Python 3.x

Python 3 is installed and ready. Any Python 3 version will work for this project.

ⓧ Command not found

Python 3 is not installed. Visit python.org/downloads and download the latest Python 3 installer for your operating system.

  • Run the installer and follow the prompts.
  • On Windows, make sure to check Add Python to PATH during installation.
  • Close and reopen your terminal, then run python3 --version again to confirm.
  • Install the boto3 library by running:
pip install boto3

What is boto3?

boto3 is the official AWS SDK for Python. It translates your Python function calls into API requests that AWS services understand.

Every script in this project uses boto3 to create, read from, and delete Kinesis streams and SQS queues.

You should see a success message ending with "Successfully installed boto3" and its dependencies.

Getting a permission error?

If you see a permission denied error, try running pip install boto3 --user instead. This installs the package to your user directory without needing admin access.

If pip is not found, try pip3 install boto3 instead.

help me install boto3 with pip

Configure AWS CLI credentials

The AWS CLI needs credentials to authenticate your API requests. You will connect it to an IAM user that has permission to create and manage Kinesis streams and SQS queues.

  • Start the configuration wizard by running:
aws configure

The wizard will prompt you for four values, one at a time.

  • Enter your AWS Access Key ID when prompted.
  • Enter your AWS Secret Access Key when prompted.
  • Enter us-east-1 for the default region.
  • Press Enter to accept the default output format (json).

Where do I find my access keys?

Access keys come from an IAM user in your AWS account. If you do not have one yet, sign into the AWS Console, navigate to IAM, create a user with programmatic access, and attach the AdministratorAccess policy for this learning project.

In production you would scope permissions down to only Kinesis and SQS actions. For a short-lived learning exercise, admin access avoids permission debugging.

  • Verify your configuration is working by running:
aws sts get-caller-identity

You should see a JSON response showing your account ID, user ARN, and user ID. This confirms the CLI can authenticate with AWS.

Seeing an InvalidClientTokenId error?

Double-check that you pasted the access key ID and secret without any extra spaces. Run aws configure again to re-enter them.

Also confirm your IAM user is active and not deleted in the AWS Console.

help me fix my AWS CLI credentials

Your environment is ready. Next up, you will create the Kinesis stream and SQS queue that form the backbone of your pipeline.

Create Your AWS Resources

Your environment is configured and ready to talk to AWS. Now it's time to create the two core resources your pipeline depends on.

You need a Kinesis Data Stream to ingest ordered events, and an SQS queue to dispatch independent tasks. In this step, you will write a Python script that provisions both resources in a single run.

In this step, get ready to:
  • Write a Python script that creates a Kinesis stream and an SQS queue.
  • Run the script and verify both resources are active.
  • Observe the cost difference between Kinesis and SQS.
Write the resource creation script

You will create a single script called create_resources.py that handles provisioning both AWS resources. This keeps your infrastructure setup repeatable and easy to tear down later.

  • Create a new file called create_resources.py in your project folder.
  • Add the following imports and configuration at the top of the file:
import boto3
import time
import json

STREAM_NAME = "order-events"
QUEUE_NAME = "order-tasks"
REGION = "us-east-1"

What does this code do?

These lines import the libraries you need and define three constants that the rest of the script uses.

  • boto3 is the AWS SDK for Python. It lets you create and manage AWS resources from code.
  • STREAM_NAME and QUEUE_NAME define the names of the resources you are about to create.
  • REGION pins everything to us-east-1 so that your credentials and resources stay in the same region.
  • Below the constants, add the function that creates your Kinesis Data Stream:
def create_kinesis_stream(kinesis_client):
    """Create a Kinesis Data Stream with 1 shard in provisioned mode."""
    print(f"Creating Kinesis stream '{STREAM_NAME}'...")
    kinesis_client.create_stream(
        StreamName=STREAM_NAME,
        ShardCount=1,
        StreamModeDetails={"StreamMode": "PROVISIONED"}
    )

    # Wait for stream to become ACTIVE
    print("Waiting for stream to become ACTIVE...")
    waiter = kinesis_client.get_waiter("stream_exists")
    waiter.wait(StreamName=STREAM_NAME)
    print(f"Stream '{STREAM_NAME}' is ACTIVE.")

What does this function do?

This function calls the Kinesis API to create a stream named order-events with 1 shard in provisioned mode.

  • ShardCount=1 means the stream has one shard. A shard is a unit of read/write capacity. One shard supports up to 1,000 records per second for writes.
  • StreamModeDetails={"StreamMode": "PROVISIONED"} tells AWS you want to manage capacity manually. This is the cheapest option for a learning exercise.
  • The waiter blocks the script until Kinesis confirms the stream is ACTIVE. Creating a stream is asynchronous, so without the waiter you could try to use it before it is ready.
  • Below the Kinesis function, add the function that creates your SQS queue:
def create_sqs_queue(sqs_client):
    """Create a standard SQS queue."""
    print(f"Creating SQS queue '{QUEUE_NAME}'...")
    response = sqs_client.create_queue(QueueName=QUEUE_NAME)
    queue_url = response["QueueUrl"]
    print(f"Queue created: {queue_url}")
    return queue_url

What does this function do?

This function creates a Standard SQS queue named order-tasks. Standard queues offer high throughput and at-least-once delivery, but messages may arrive out of order.

The QueueUrl returned by AWS is the address you will use to send and receive messages later. SQS queue creation is synchronous, so there is no waiter needed here.

  • Finally, add the main function and the entry point at the bottom of the file:
def main():
    kinesis_client = boto3.client("kinesis", region_name=REGION)
    sqs_client = boto3.client("sqs", region_name=REGION)

    create_kinesis_stream(kinesis_client)
    queue_url = create_sqs_queue(sqs_client)

    print("\n--- Resources Created ---")
    print(f"Kinesis Stream: {STREAM_NAME} (1 shard, provisioned)")
    print(f"SQS Queue URL: {queue_url}")
    print("\n⚠️  Kinesis costs $0.015/hour. Remember to delete when done!")


if __name__ == "__main__":
    main()

What does this code do?

  • boto3.client("kinesis", region_name=REGION) creates a low-level client for talking to Kinesis.
  • boto3.client("sqs", region_name=REGION) creates a client for talking to SQS.
  • The main function calls both creation helpers in sequence and prints a summary when both resources are ready.
  • The if __name__ == "__main__" guard ensures main() only runs when you execute the script directly.
  • Save create_resources.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

import boto3
import time
import json

STREAM_NAME = "order-events"
QUEUE_NAME = "order-tasks"
REGION = "us-east-1"

def create_kinesis_stream(kinesis_client):
    """Create a Kinesis Data Stream with 1 shard in provisioned mode."""
    print(f"Creating Kinesis stream '{STREAM_NAME}'...")
    kinesis_client.create_stream(
        StreamName=STREAM_NAME,
        ShardCount=1,
        StreamModeDetails={"StreamMode": "PROVISIONED"}
    )

    # Wait for stream to become ACTIVE
    print("Waiting for stream to become ACTIVE...")
    waiter = kinesis_client.get_waiter("stream_exists")
    waiter.wait(StreamName=STREAM_NAME)
    print(f"Stream '{STREAM_NAME}' is ACTIVE.")


def create_sqs_queue(sqs_client):
    """Create a standard SQS queue."""
    print(f"Creating SQS queue '{QUEUE_NAME}'...")
    response = sqs_client.create_queue(QueueName=QUEUE_NAME)
    queue_url = response["QueueUrl"]
    print(f"Queue created: {queue_url}")
    return queue_url


def main():
    kinesis_client = boto3.client("kinesis", region_name=REGION)
    sqs_client = boto3.client("sqs", region_name=REGION)

    create_kinesis_stream(kinesis_client)
    queue_url = create_sqs_queue(sqs_client)

    print("\n--- Resources Created ---")
    print(f"Kinesis Stream: {STREAM_NAME} (1 shard, provisioned)")
    print(f"SQS Queue URL: {queue_url}")
    print("\n⚠️  Kinesis costs $0.015/hour. Remember to delete when done!")


if __name__ == "__main__":
    main()
Run the script and verify your resources

The script handles everything: creating the stream, waiting for it to become active, and then creating the queue. One command and both resources will be live.

  • Run the script by executing this command in your terminal:
python3 create_resources.py
  • Wait for the script to finish. The Kinesis stream takes 10-20 seconds to become active.
  • You should see output similar to this:
Creating Kinesis stream 'order-events'...
Waiting for stream to become ACTIVE...
Stream 'order-events' is ACTIVE.
Creating SQS queue 'order-tasks'...
Queue created: https://sqs.us-east-1.amazonaws.com/123456789012/order-tasks

--- Resources Created ---
Kinesis Stream: order-events (1 shard, provisioned)
SQS Queue URL: https://sqs.us-east-1.amazonaws.com/123456789012/order-tasks

⚠️  Kinesis costs $0.015/hour. Remember to delete when done!

What should I see?

The key confirmation is the line that says Stream 'order-events' is ACTIVE. followed by Queue created: with a URL. The URL contains your AWS account number, so it will differ from the example above.

Seeing a ResourceInUseException?

This error means a stream named order-events already exists in your account in us-east-1. If you ran the script before or created one manually, delete the existing stream first, wait a minute for deletion to complete, then re-run the script.

If you see a credentials or permissions error, double check that the IAM user you configured with aws configure has Kinesis and SQS permissions.

Help me debug my create_resources.py script

Observe the cost difference

Both resources are now live in your AWS account. Their cost profiles are very different, and this difference is part of why architects use them for different purposes.

How much does this cost?

  • Kinesis Data Streams is NOT in the AWS Free Tier. Your 1-shard stream costs $0.015 per hour starting right now. That works out to $0.36 per day if you forget to delete it.
  • SQS gives you 1 million requests per month for free. This project uses fewer than 100 requests total, so your SQS usage costs nothing.

This cost asymmetry reflects what each service provides. Kinesis reserves dedicated shard capacity for ordered, real-time ingestion. SQS charges only when you actually send and receive messages.

Remember to clean up

You will delete both resources in a later step. If you need to stop partway through this project, run the delete_resources.py script (covered in Step 6) to avoid ongoing charges.

Your stream and queue are live. Next up, you will write a producer that sends simulated order events into Kinesis and see how partition keys control ordering.

Produce Order Events into Kinesis

Your Kinesis stream and SQS queue are live and waiting for data. Now it's time to simulate real order events flowing into the stream.

In this step, you'll write a producer script that generates order events and pushes them into your Kinesis Data Stream. The critical detail is the partition key, which determines how records get routed to shards and, ultimately, how ordering is guaranteed.

In this step, get ready to:
  • Write producer.py to generate and send order events to Kinesis.
  • Run the producer and observe shard routing in the output.
  • Understand why Kinesis ordering matters for event streams.
Write the producer script

The producer is the entry point for data in any streaming architecture. It simulates e-commerce orders from multiple customers and sends each one to the Kinesis stream as a JSON record.

  • Create a new file called producer.py in the same folder as create_resources.py.

You'll build this file in three chunks.

  • Add the imports and configuration constants to producer.py by pasting this code:
import boto3
import json
import time
import uuid
from datetime import datetime, timezone

STREAM_NAME = "order-events"
REGION = "us-east-1"

CUSTOMERS = ["customer-A", "customer-B", "customer-C"]
ITEMS = ["laptop", "headphones", "keyboard", "monitor", "mouse"]

What does this code do?

STREAM_NAME matches the exact name you created in create_resources.py, so the producer knows which stream to target.

CUSTOMERS and ITEMS are lists of simulated data. The script cycles through these to create varied but predictable order events.

  • Add the generate_order function below the constants:

def generate_order(customer_id, item):
    """Generate a simulated order event."""
    return {
        "order_id": str(uuid.uuid4())[:8],
        "customer_id": customer_id,
        "item": item,
        "quantity": 1,
        "timestamp": datetime.now(timezone.utc).isoformat()
    }

What does this code do?

This function builds a dictionary representing one order event. It generates a short unique order_id using uuid4(), includes the customer and item, and timestamps the event in UTC.

  • Add the produce_orders function and the main block below the generate_order function:

def produce_orders(kinesis_client, num_orders=10):
    """Produce order events into the Kinesis stream."""
    print(f"Producing {num_orders} orders into '{STREAM_NAME}'...\n")

    for i in range(num_orders):
        customer_id = CUSTOMERS[i % len(CUSTOMERS)]
        item = ITEMS[i % len(ITEMS)]
        order = generate_order(customer_id, item)

        response = kinesis_client.put_record(
            StreamName=STREAM_NAME,
            Data=json.dumps(order),
            PartitionKey=customer_id
        )

        print(f"Order {i+1}: {order['order_id']} | "
              f"Customer: {customer_id} | "
              f"Item: {item} | "
              f"Shard: {response['ShardId']} | "
              f"Seq: {response['SequenceNumber'][:12]}...")

        time.sleep(0.5)

    print(f"\nDone! Produced {num_orders} orders.")
    print("Notice: All records go to the same shard (we only have 1).")
    print("With multiple shards, same partition key = same shard = ordering preserved.")


def main():
    kinesis_client = boto3.client("kinesis", region_name=REGION)
    produce_orders(kinesis_client)


if __name__ == "__main__":
    main()

What does this code do?

  • put_record sends a single record to the Kinesis stream. It takes the stream name, the JSON-encoded order data, and a partition key.
  • PartitionKey=customer_id is the critical line. Kinesis uses an MD5 hash of this key to decide which shard receives the record.
  • The response from put_record returns the ShardId (which shard stored the record) and a SequenceNumber (a unique, increasing identifier for the record within that shard).
  • time.sleep(0.5) adds a half-second pause between records so you can watch the output scroll by in real time.
  • Save producer.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before running it.

ⓧ I'd like to double check the full code

import boto3
import json
import time
import uuid
from datetime import datetime, timezone

STREAM_NAME = "order-events"
REGION = "us-east-1"

CUSTOMERS = ["customer-A", "customer-B", "customer-C"]
ITEMS = ["laptop", "headphones", "keyboard", "monitor", "mouse"]


def generate_order(customer_id, item):
    """Generate a simulated order event."""
    return {
        "order_id": str(uuid.uuid4())[:8],
        "customer_id": customer_id,
        "item": item,
        "quantity": 1,
        "timestamp": datetime.now(timezone.utc).isoformat()
    }


def produce_orders(kinesis_client, num_orders=10):
    """Produce order events into the Kinesis stream."""
    print(f"Producing {num_orders} orders into '{STREAM_NAME}'...\n")

    for i in range(num_orders):
        customer_id = CUSTOMERS[i % len(CUSTOMERS)]
        item = ITEMS[i % len(ITEMS)]
        order = generate_order(customer_id, item)

        response = kinesis_client.put_record(
            StreamName=STREAM_NAME,
            Data=json.dumps(order),
            PartitionKey=customer_id
        )

        print(f"Order {i+1}: {order['order_id']} | "
              f"Customer: {customer_id} | "
              f"Item: {item} | "
              f"Shard: {response['ShardId']} | "
              f"Seq: {response['SequenceNumber'][:12]}...")

        time.sleep(0.5)

    print(f"\nDone! Produced {num_orders} orders.")
    print("Notice: All records go to the same shard (we only have 1).")
    print("With multiple shards, same partition key = same shard = ordering preserved.")


def main():
    kinesis_client = boto3.client("kinesis", region_name=REGION)
    produce_orders(kinesis_client)


if __name__ == "__main__":
    main()
Run the producer and observe shard routing
  • Run the producer script by executing this command:
python3 producer.py

You should see 10 lines of output, one per order, each showing the order ID, customer, item, shard ID, and a truncated sequence number.

What should I see?

Each line confirms that Kinesis accepted the record. The Shard column shows shardId-000000000000 for every record because your stream only has one shard.

The Seq column shows a truncated sequence number. Notice that these numbers increase with each record. Sequence numbers are unique within a shard and always increase over time.

Seeing a ResourceNotFoundException?

This means the Kinesis stream order-events does not exist. Make sure you ran create_resources.py successfully in the previous step and that the stream reached ACTIVE status.

Also check that the REGION in producer.py matches the region you configured with aws configure.

help me debug a ResourceNotFoundException when calling put_record on my Kinesis stream

Why use customer_id as the partition key?

Kinesis uses an MD5 hash of the partition key to determine which shard receives each record. All records with the same partition key land on the same shard.

With one shard, this detail is invisible. With multiple shards, it means all orders from customer-A always land on the same shard. A consumer reading that shard sees customer-A's orders in exactly the order they were placed.

Understand the Kinesis ordering guarantee

Take a moment to reflect on what just happened. You sent 10 orders from 3 different customers into a single shard.

Why does ordering matter here?

Imagine customer-A places order 1 and then order 2. Because both records share the same partition key, they land on the same shard. Within that shard, records are strictly ordered by sequence number.

Any consumer reading that shard will always see order 1 before order 2. This guarantee is what makes Kinesis the right tool for event streams where sequence matters.

If order processing happened out of sequence (shipping before payment confirmation, for example), your system would break. Kinesis prevents that by design.

Your stream now holds 10 ordered records. Next up, you'll consume these records from Kinesis and fan out tasks to your SQS queue.

Consume from Kinesis and Dispatch to SQS

Your Kinesis stream is loaded with 10 order events, sitting in sequence order inside the shard. Now you need something to read those events and turn each one into actionable work.

In this step, you will write a processor that reads from Kinesis in order and dispatches individual tasks to your SQS queue. This is the bridge between the ordered event stream and the decoupled task queue.

In this step, get ready to:
  • Write a processor that reads Kinesis records using a shard iterator.
  • Dispatch tasks to SQS for each order event.
  • Observe the Kinesis ordering guarantee in action.
Write the processor script

The processor connects both AWS services together. It reads events from Kinesis in sequence order, then for each event it sends discrete tasks to SQS.

  • Create a new file called processor.py in your project folder.
  • Add the imports, constants, and a helper function to look up the SQS queue URL:
import boto3
import json
import time

STREAM_NAME = "order-events"
QUEUE_NAME = "order-tasks"
REGION = "us-east-1"


def get_queue_url(sqs_client):
    """Get the URL for our SQS queue."""
    response = sqs_client.get_queue_url(QueueName=QUEUE_NAME)
    return response["QueueUrl"]

What does this code do?

The constants match the resource names you created in create_resources.py.

The get_queue_url function looks up the full URL for your SQS queue by name. SQS operations require the queue URL, not just the name.

  • Below get_queue_url, add the dispatch_tasks function that creates SQS messages for each order:
def dispatch_tasks(sqs_client, queue_url, order):
    """For each order, dispatch tasks to SQS."""
    tasks = [
        {"task_type": "send_confirmation", "order_id": order["order_id"], "customer_id": order["customer_id"]},
        {"task_type": "update_inventory", "order_id": order["order_id"], "item": order["item"]},
    ]

    for task in tasks:
        sqs_client.send_message(
            QueueUrl=queue_url,
            MessageBody=json.dumps(task)
        )
        print(f"  -> Dispatched task: {task['task_type']} for order {task['order_id']}")

What does this code do?

For every single order event, this function creates two separate tasks: send_confirmation and update_inventory.

Each task is sent as a JSON message to the SQS queue using send_message. This is the "fanout" pattern. One event becomes multiple independent tasks.

  • Below dispatch_tasks, add the consume_stream function that reads records from the Kinesis shard:
def consume_stream(kinesis_client, sqs_client, queue_url):
    """Read all records from the Kinesis stream and dispatch tasks to SQS."""
    # Describe stream to get shard info
    response = kinesis_client.describe_stream(StreamName=STREAM_NAME)
    shard_id = response["StreamDescription"]["Shards"][0]["ShardId"]

    # Get shard iterator from the beginning (TRIM_HORIZON)
    iterator_response = kinesis_client.get_shard_iterator(
        StreamName=STREAM_NAME,
        ShardId=shard_id,
        ShardIteratorType="TRIM_HORIZON"
    )
    shard_iterator = iterator_response["ShardIterator"]

    print(f"Reading from Kinesis stream '{STREAM_NAME}' (shard: {shard_id})...")
    print("Records arrive in sequence order. This is the Kinesis ordering guarantee.\n")

    total_records = 0
    empty_reads = 0

    while empty_reads < 3:
        records_response = kinesis_client.get_records(
            ShardIterator=shard_iterator,
            Limit=10
        )

        records = records_response["Records"]
        shard_iterator = records_response["NextShardIterator"]

        if not records:
            empty_reads += 1
            time.sleep(1)
            continue

        empty_reads = 0

        for record in records:
            total_records += 1
            order = json.loads(record["Data"])
            seq = record["SequenceNumber"]

            print(f"[Record #{total_records}] Seq: {seq[:12]}... | "
                  f"Order: {order['order_id']} | "
                  f"Customer: {order['customer_id']} | "
                  f"Item: {order['item']}")

            dispatch_tasks(sqs_client, queue_url, order)
            print()

        time.sleep(0.5)

    print(f"\nDone! Processed {total_records} records from Kinesis.")
    print(f"Dispatched {total_records * 2} tasks to SQS queue '{QUEUE_NAME}'.")

What does this code do?

  • describe_stream fetches shard information so the consumer knows which shard to read from.
  • get_shard_iterator with TRIM_HORIZON starts reading from the oldest record in the shard.
  • get_records pulls a batch of records and returns a new iterator for the next batch.
  • The loop keeps reading until it gets 3 consecutive empty responses, meaning it has caught up with the end of the stream.
  • For each record, it decodes the JSON payload, prints the sequence number to prove ordering, then calls dispatch_tasks to fan out work to SQS.

What is TRIM_HORIZON?

TRIM_HORIZON tells Kinesis to start reading from the oldest available record in the shard. The alternative is LATEST, which starts from new records arriving after the iterator is created.

Since your producer already wrote 10 records, you need TRIM_HORIZON to go back and read them all from the beginning.

  • Finally, add the main function and entry point at the bottom of processor.py:
def main():
    kinesis_client = boto3.client("kinesis", region_name=REGION)
    sqs_client = boto3.client("sqs", region_name=REGION)

    queue_url = get_queue_url(sqs_client)
    consume_stream(kinesis_client, sqs_client, queue_url)


if __name__ == "__main__":
    main()

What does this code do?

The main function creates both a Kinesis client and an SQS client, looks up the queue URL, then kicks off the stream consumption loop.

  • Save processor.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before running it in the next substep.

ⓧ I'd like to double check the full code

import boto3
import json
import time

STREAM_NAME = "order-events"
QUEUE_NAME = "order-tasks"
REGION = "us-east-1"


def get_queue_url(sqs_client):
    """Get the URL for our SQS queue."""
    response = sqs_client.get_queue_url(QueueName=QUEUE_NAME)
    return response["QueueUrl"]


def dispatch_tasks(sqs_client, queue_url, order):
    """For each order, dispatch tasks to SQS."""
    tasks = [
        {"task_type": "send_confirmation", "order_id": order["order_id"], "customer_id": order["customer_id"]},
        {"task_type": "update_inventory", "order_id": order["order_id"], "item": order["item"]},
    ]

    for task in tasks:
        sqs_client.send_message(
            QueueUrl=queue_url,
            MessageBody=json.dumps(task)
        )
        print(f"  -> Dispatched task: {task['task_type']} for order {task['order_id']}")


def consume_stream(kinesis_client, sqs_client, queue_url):
    """Read all records from the Kinesis stream and dispatch tasks to SQS."""
    # Describe stream to get shard info
    response = kinesis_client.describe_stream(StreamName=STREAM_NAME)
    shard_id = response["StreamDescription"]["Shards"][0]["ShardId"]

    # Get shard iterator from the beginning (TRIM_HORIZON)
    iterator_response = kinesis_client.get_shard_iterator(
        StreamName=STREAM_NAME,
        ShardId=shard_id,
        ShardIteratorType="TRIM_HORIZON"
    )
    shard_iterator = iterator_response["ShardIterator"]

    print(f"Reading from Kinesis stream '{STREAM_NAME}' (shard: {shard_id})...")
    print("Records arrive in sequence order. This is the Kinesis ordering guarantee.\n")

    total_records = 0
    empty_reads = 0

    while empty_reads < 3:
        records_response = kinesis_client.get_records(
            ShardIterator=shard_iterator,
            Limit=10
        )

        records = records_response["Records"]
        shard_iterator = records_response["NextShardIterator"]

        if not records:
            empty_reads += 1
            time.sleep(1)
            continue

        empty_reads = 0

        for record in records:
            total_records += 1
            order = json.loads(record["Data"])
            seq = record["SequenceNumber"]

            print(f"[Record #{total_records}] Seq: {seq[:12]}... | "
                  f"Order: {order['order_id']} | "
                  f"Customer: {order['customer_id']} | "
                  f"Item: {order['item']}")

            dispatch_tasks(sqs_client, queue_url, order)
            print()

        time.sleep(0.5)

    print(f"\nDone! Processed {total_records} records from Kinesis.")
    print(f"Dispatched {total_records * 2} tasks to SQS queue '{QUEUE_NAME}'.")


def main():
    kinesis_client = boto3.client("kinesis", region_name=REGION)
    sqs_client = boto3.client("sqs", region_name=REGION)

    queue_url = get_queue_url(sqs_client)
    consume_stream(kinesis_client, sqs_client, queue_url)


if __name__ == "__main__":
    main()
Run the processor and observe ordering
  • Run the processor by executing this command:
python3 processor.py

The output shows each record with its sequence number, proving that Kinesis delivered them in the exact order they were produced. After each record, you see two dispatched tasks sent to SQS.

What should I see?

You should see 10 records printed in ascending sequence number order. Each record shows its order ID, customer, and item.

Below each record, two lines confirm that send_confirmation and update_inventory tasks were dispatched to SQS. The final summary should say 10 records processed and 20 tasks dispatched.

Seeing zero records?

If the processor reports 0 records, your data may have expired from the stream. Kinesis has a default 24-hour retention period, so records you produced earlier today should still be available.

  • Make sure you ran producer.py successfully in the previous step.
  • Confirm the stream name order-events matches in both scripts.
  • Check that your region is set to us-east-1 in both files.

help me debug why my processor is reading zero records from Kinesis

Understand why SQS handles the fanout

Look at what just happened from an architecture perspective. The processor read events from Kinesis in strict sequence order.

Why use SQS for the fanout?

The tasks dispatched to SQS do not need ordering. "Send email for order 5" and "update inventory for order 3" can be handled in any order, by any number of workers, at any pace.

SQS gives you reliable delivery and decoupling, not ordering. If a worker crashes, the message stays in the queue and another worker can pick it up. You can add more workers without changing any code.

This is why production systems use both services together. Kinesis preserves the event timeline. SQS handles the independent work that comes out of it.

Your processor has proven the core pattern: ordered ingestion through Kinesis, decoupled task delivery through SQS. Next up, you will write a worker that polls the SQS queue and handles those tasks independently.

Process Tasks from SQS

Your processor just read every order event from Kinesis in perfect sequence and dispatched 20 tasks into the SQS queue. Now it's time to build the other side of the pipeline: a worker that picks up those tasks and handles them.

The worker's job is simple. It polls the queue, processes each task, then deletes it so no other worker picks it up again. The interesting part is what you'll observe about message ordering.

In this step, get ready to:
  • Write a worker script that polls SQS using long polling and processes tasks.
  • Run the worker and observe that tasks arrive in a different order than they were sent.
  • Contrast the ordering guarantees of Kinesis vs the delivery behavior of SQS.
Write the SQS worker script

The worker is the final piece of your pipeline. It continuously polls the SQS queue for available messages, handles each one, then deletes it from the queue to confirm successful processing.

  • Create a new file called worker.py in the same folder as your other scripts.
  • Add the imports, constants, and queue URL helper by pasting this code:
import boto3
import json
import time

QUEUE_NAME = "order-tasks"
REGION = "us-east-1"


def get_queue_url(sqs_client):
    """Get the URL for our SQS queue."""
    response = sqs_client.get_queue_url(QueueName=QUEUE_NAME)
    return response["QueueUrl"]

What does this code do?

This sets up the same pattern you used in processor.py. The get_queue_url function looks up the live URL for the order-tasks queue by name, so your code doesn't need to hardcode the URL.

  • Next, add the task processing function below get_queue_url:

def process_task(task):
    """Simulate processing a task."""
    if task["task_type"] == "send_confirmation":
        print(f"  📧 Sending confirmation email for order {task['order_id']} "
              f"to {task['customer_id']}")
    elif task["task_type"] == "update_inventory":
        print(f"  📦 Updating inventory: deducting {task['item']} "
              f"for order {task['order_id']}")
    else:
        print(f"  ❓ Unknown task type: {task['task_type']}")

What does this code do?

This function simulates what a real worker would do. In production, the send_confirmation branch would call an email API, and update_inventory would write to a database.

For this project, printing the task details is enough to prove the messages are flowing correctly.

  • Now add the main polling loop below process_task:

def poll_queue(sqs_client, queue_url):
    """Poll SQS queue and process tasks."""
    print(f"Polling SQS queue '{QUEUE_NAME}' for tasks...")
    print("(Tasks may arrive in any order. That's fine! They're independent.)\n")

    total_processed = 0
    empty_polls = 0

    while empty_polls < 3:
        response = sqs_client.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=5
        )

        messages = response.get("Messages", [])

        if not messages:
            empty_polls += 1
            print(f"No messages received (poll {empty_polls}/3)...")
            continue

        empty_polls = 0

        for message in messages:
            total_processed += 1
            task = json.loads(message["Body"])

            print(f"[Task #{total_processed}] {task['task_type']} | order: {task['order_id']}")
            process_task(task)

            # Delete message after successful processing
            sqs_client.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message["ReceiptHandle"]
            )
            print(f"  ✓ Deleted from queue\n")

    print(f"\nDone! Processed {total_processed} tasks from SQS.")
    print("Notice: tasks arrived in a different order than they were sent.")
    print("This is expected with Standard SQS, and it's perfectly fine for independent tasks.")

What does this code do?

  • WaitTimeSeconds=5 enables long polling. Instead of returning immediately when the queue is empty, SQS waits up to 5 seconds for a message to arrive before returning an empty response. This reduces unnecessary API calls.
  • MaxNumberOfMessages=10 pulls up to 10 messages per request, which is more efficient than fetching one at a time.
  • delete_message removes the message from the queue after processing. If you skip this step, the message reappears after the visibility timeout expires, which could cause duplicate processing.
  • The loop exits after 3 consecutive empty polls, meaning the queue has been fully drained.
  • Finally, add the entry point at the bottom of the file:

def main():
    sqs_client = boto3.client("sqs", region_name=REGION)
    queue_url = get_queue_url(sqs_client)
    poll_queue(sqs_client, queue_url)


if __name__ == "__main__":
    main()

What does this code do?

This follows the same pattern as your other scripts. It creates an SQS client, looks up the queue URL, and starts polling.

  • Save worker.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before running it.

ⓧ I'd like to double check the full code

import boto3
import json
import time

QUEUE_NAME = "order-tasks"
REGION = "us-east-1"


def get_queue_url(sqs_client):
    """Get the URL for our SQS queue."""
    response = sqs_client.get_queue_url(QueueName=QUEUE_NAME)
    return response["QueueUrl"]


def process_task(task):
    """Simulate processing a task."""
    if task["task_type"] == "send_confirmation":
        print(f"  📧 Sending confirmation email for order {task['order_id']} "
              f"to {task['customer_id']}")
    elif task["task_type"] == "update_inventory":
        print(f"  📦 Updating inventory: deducting {task['item']} "
              f"for order {task['order_id']}")
    else:
        print(f"  ❓ Unknown task type: {task['task_type']}")


def poll_queue(sqs_client, queue_url):
    """Poll SQS queue and process tasks."""
    print(f"Polling SQS queue '{QUEUE_NAME}' for tasks...")
    print("(Tasks may arrive in any order. That's fine! They're independent.)\n")

    total_processed = 0
    empty_polls = 0

    while empty_polls < 3:
        response = sqs_client.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=5
        )

        messages = response.get("Messages", [])

        if not messages:
            empty_polls += 1
            print(f"No messages received (poll {empty_polls}/3)...")
            continue

        empty_polls = 0

        for message in messages:
            total_processed += 1
            task = json.loads(message["Body"])

            print(f"[Task #{total_processed}] {task['task_type']} | order: {task['order_id']}")
            process_task(task)

            # Delete message after successful processing
            sqs_client.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message["ReceiptHandle"]
            )
            print(f"  ✓ Deleted from queue\n")

    print(f"\nDone! Processed {total_processed} tasks from SQS.")
    print("Notice: tasks arrived in a different order than they were sent.")
    print("This is expected with Standard SQS, and it's perfectly fine for independent tasks.")


def main():
    sqs_client = boto3.client("sqs", region_name=REGION)
    queue_url = get_queue_url(sqs_client)
    poll_queue(sqs_client, queue_url)


if __name__ == "__main__":
    main()
Run the worker and observe task consumption

Your SQS queue currently holds 20 task messages from the processor. The worker will drain all of them.

  • Run the worker by executing this command in your terminal:
python3 worker.py

You should see the worker pick up tasks in batches, printing the task type and order ID for each one. After all 20 tasks are processed and deleted, the worker exits after 3 empty polls.

What should I see?

Your terminal shows lines like:

[Task #1] send_confirmation | order: a3f2b1c8
  📧 Sending confirmation email for order a3f2b1c8 to customer-B
  ✓ Deleted from queue

[Task #2] update_inventory | order: 7e4d9a12
  📦 Updating inventory: deducting keyboard for order 7e4d9a12
  ✓ Deleted from queue

The final summary says something like "Done! Processed 20 tasks from SQS" followed by a note about ordering. Your specific order IDs will differ from the example above.

Not seeing any messages?

  • Make sure you ran processor.py in Step 4 before running this worker. The processor is what sends messages into the queue.
  • If you already ran the worker once, the queue is now empty. Run processor.py again to refill the queue, then run worker.py again.
  • Check that QUEUE_NAME in worker.py matches the queue name in create_resources.py (both should be order-tasks).

Still stuck? Help me debug why my SQS worker isn't receiving messages.

Contrast Kinesis ordering vs SQS delivery

Look at the order in which your worker received tasks. Compare it to the order in which the processor dispatched them.

Why did the tasks arrive out of order?

Standard SQS queues provide best-effort ordering. Messages may arrive in a different sequence than they were sent. This is a deliberate design tradeoff for higher throughput and availability.

For your pipeline, this is perfectly fine. "Send confirmation email for order 5" and "update inventory for order 3" are completely independent tasks. It doesn't matter which one executes first.

When to use Kinesis vs SQS

Kinesis gave you ordered delivery. Every record within a shard arrives in the exact sequence it was produced. This is critical for the event stream where customer A's order 1 must be processed before order 2.

SQS gave you decoupled, reliable task delivery. Each message is processed exactly once (via delete-after-success), and workers can scale independently without coordinating with the producer.

Each service did what it was built for. Production systems use both together for exactly this reason.

Your pipeline is complete. You've proven when ordering matters (Kinesis) and when decoupled delivery is the better fit (SQS). Next up, you'll clean up both resources so that Kinesis shard doesn't keep billing you.

Clean Up Your AWS Resources

Your pipeline works perfectly. But that Kinesis stream is still running and costing $0.015 per hour while it exists.

In this step, you'll write a cleanup script that deletes both your Kinesis Data Stream and your SQS queue. Once you run it, charges stop immediately.

In this step, get ready to:
  • Write delete_resources.py to remove your AWS resources.
  • Run the script and verify both resources are deleted.
Write and run the cleanup script

The cleanup script uses the same boto3 clients you've been working with, but calls the delete APIs instead of create.

  • Create a new file called delete_resources.py in your project folder.
  • Paste the following code into the file:
import boto3
import time

STREAM_NAME = "order-events"
QUEUE_NAME = "order-tasks"
REGION = "us-east-1"


def delete_kinesis_stream(kinesis_client):
    """Delete the Kinesis Data Stream."""
    print(f"Deleting Kinesis stream '{STREAM_NAME}'...")
    kinesis_client.delete_stream(StreamName=STREAM_NAME, EnforceConsumerDeletion=True)
    print(f"Stream '{STREAM_NAME}' deletion initiated.")
    print("(Stream will be fully removed in a few seconds.)")


def delete_sqs_queue(sqs_client):
    """Delete the SQS queue."""
    print(f"Deleting SQS queue '{QUEUE_NAME}'...")
    response = sqs_client.get_queue_url(QueueName=QUEUE_NAME)
    queue_url = response["QueueUrl"]
    sqs_client.delete_queue(QueueUrl=queue_url)
    print(f"Queue '{QUEUE_NAME}' deleted.")


def main():
    kinesis_client = boto3.client("kinesis", region_name=REGION)
    sqs_client = boto3.client("sqs", region_name=REGION)

    delete_kinesis_stream(kinesis_client)
    delete_sqs_queue(sqs_client)

    print("\n--- All Resources Deleted ---")
    print("No more charges will accrue. You're clean!")


if __name__ == "__main__":
    main()

What does this code do?

  • delete_kinesis_stream() calls delete_stream with EnforceConsumerDeletion=True to delete the stream even if consumers are still registered.
  • delete_sqs_queue() first looks up the queue URL by name, then calls delete_queue to remove it permanently.
  • The main() function creates both boto3 clients and runs both deletions in sequence.
  • Save delete_resources.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before running it.

ⓧ I'd like to double check the full code

Here is the complete delete_resources.py file:

import boto3
import time

STREAM_NAME = "order-events"
QUEUE_NAME = "order-tasks"
REGION = "us-east-1"


def delete_kinesis_stream(kinesis_client):
    """Delete the Kinesis Data Stream."""
    print(f"Deleting Kinesis stream '{STREAM_NAME}'...")
    kinesis_client.delete_stream(StreamName=STREAM_NAME, EnforceConsumerDeletion=True)
    print(f"Stream '{STREAM_NAME}' deletion initiated.")
    print("(Stream will be fully removed in a few seconds.)")


def delete_sqs_queue(sqs_client):
    """Delete the SQS queue."""
    print(f"Deleting SQS queue '{QUEUE_NAME}'...")
    response = sqs_client.get_queue_url(QueueName=QUEUE_NAME)
    queue_url = response["QueueUrl"]
    sqs_client.delete_queue(QueueUrl=queue_url)
    print(f"Queue '{QUEUE_NAME}' deleted.")


def main():
    kinesis_client = boto3.client("kinesis", region_name=REGION)
    sqs_client = boto3.client("sqs", region_name=REGION)

    delete_kinesis_stream(kinesis_client)
    delete_sqs_queue(sqs_client)

    print("\n--- All Resources Deleted ---")
    print("No more charges will accrue. You're clean!")


if __name__ == "__main__":
    main()
  • Run the cleanup script by executing this command:
python delete_resources.py

You should see output confirming both deletions, ending with "All Resources Deleted" and "No more charges will accrue."

Seeing an error?

  • If you see "ResourceNotFoundException", the stream may have already been deleted. This is fine.
  • If you see "QueueDoesNotExist", the SQS queue was already removed. This is also fine.
  • If you see "AccessDeniedException", your IAM user may not have delete permissions. Check that your credentials have Kinesis and SQS access.

Help me troubleshoot my delete_resources.py script

Verify your resources are deleted

The script's output is your confirmation. When you see the "All Resources Deleted" message, both the Kinesis stream and the SQS queue are gone.

What happens after deletion?

The Kinesis stream deletion is initiated immediately, but the stream takes a few seconds to fully disappear from the AWS console. The SQS queue is deleted instantly.

Once both resources are removed, no further charges accrue. Your $0.015/hour shard cost stops the moment the deletion is initiated.

Why a script instead of the console?

You could delete these resources manually through the AWS console. But writing a deletion script means you can recreate and tear down your entire pipeline with two commands: python create_resources.py and python delete_resources.py. This is the foundation of infrastructure as code.

Secret mission

Add a Dead Letter Queue

In production, tasks fail. Servers crash, APIs time out, and bugs slip through. What happens to those failed messages? In this secret mission, you'll add a Dead Letter Queue that automatically catches messages that fail processing three times, giving you a safety net for investigating and redriving failed tasks.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Your Kinesis Data Stream costs $0.015/hour ($0.36/day) while it exists. SQS queues are free under the 1 million requests/month free tier.

Cost warning

Kinesis Data Streams is NOT in the AWS Free Tier. If you forget to delete the order-events stream, it costs $0.36/day ($10.80/month). Always clean up when you are done experimenting.

Resources you used:

  • Kinesis Data Stream order-events (1 shard, provisioned mode) — costs $0.015/hour while it exists.
  • SQS Queue order-tasks (Standard Queue with redrive policy). Free tier.
  • SQS Queue order-tasks-dlq (Dead Letter Queue). Free tier.
  • Local Python files: create_resources.py, producer.py, processor.py, worker.py, delete_resources.py, dlq_inspector.py.
  • AWS CLI credentials stored in ~/.aws/.

Keep everything running

No action needed. Choose this if you are still actively experimenting with the pipeline or plan to continue within the next few hours.

  • Your Kinesis stream continues to cost $0.015/hour ($0.36/day).
  • Your SQS queues remain free under the free tier.
  • Your local Python files stay on your machine.
  • Your AWS credentials remain configured in ~/.aws/.

Pause - I'll come back to this later

Delete the AWS resources to stop charges, but keep your local files so you can recreate everything later with a single script.

  • Run delete_resources.py to remove the Kinesis stream and both SQS queues.
  • Keep your local Python files. When you come back, run create_resources.py to recreate everything.
  • Keep your AWS credentials in ~/.aws/ so you do not need to reconfigure.

Delete the AWS resources by running this command:

python3 delete_resources.py

You should see confirmation that the stream and both queues have been deleted. No more charges will accrue.

Delete - I don't want to use this again

Remove all project resources and local files. Nothing will remain on AWS or your machine.

  • Run delete_resources.py to remove the Kinesis stream and both SQS queues.
  • Delete your local project files.
  • Optionally remove your AWS credentials.

Delete the AWS resources by running:

python3 delete_resources.py

You should see confirmation that the stream and both queues have been deleted.

Delete your local Python files by running:

rm create_resources.py producer.py processor.py worker.py delete_resources.py dlq_inspector.py

If you want to remove your AWS CLI credentials as well, run:

rm -rf ~/.aws

Removing AWS credentials

Only remove ~/.aws/ if you do not use AWS CLI for any other projects. This folder stores credentials for all AWS CLI usage on your machine.

Nice Work!

Nice Work!

You just built an event-driven order processing pipeline that uses Kinesis Data Streams and SQS together, the way production systems actually do.

You've learned how to:

  • Built a Kinesis Data Stream and produced ordered events using partition keys to guarantee sequence within a shard.
  • Consumed events in order from Kinesis and dispatched decoupled tasks to an SQS queue, proving when ordering matters and when it doesn't.
  • Built a producer/consumer pipeline with independent workers that poll, process, and delete messages at their own pace.
  • Added a Dead Letter Queue with a redrive policy for production-grade error handling and failure inspection.

Ready to quiz yourself?