streaming-03-analytics

Streaming data analytics: validate and summarize messages.

Streaming analytics requires working with data in motion and distributed, scalable systems. This course builds capabilities through working projects. In the age of generative AI, durable skills are grounded in real work: setting up a professional environment, reading and running code, understanding the logic, and pushing work to a shared repository. Each project follows the structure of professional Python projects. We learn by doing.

This Project

This project focuses on analytics performed as messages are consumed.

The project uses Kafka to move sales messages from a producer to a consumer. The consumer reads each message, validates required fields, computes derived values, writes processed records to CSV, and logs running summary statistics.

This module adds validation and message-by-message analytics to the streaming workflow.

The goal is to see how each incoming message can be checked, transformed, and summarized without waiting for a batch process.

Working Files

You'll work with just these areas:

• data/ - input data and generated output files
• docs/ - the project narrative and documentation
• src/streaming/ - producer, consumer, and supporting code
• pyproject.toml - update authorship & links
• zensical.toml - update authorship & links

Instructions

Follow the step-by-step workflow guide to complete:

1. Phase 1. Start & Run
2. Phase 2. Change Authorship
3. Phase 3. Read & Understand
4. Phase 4. Modify
5. Phase 5. Apply

Challenges

Challenges are expected. Sometimes instructions may not quite match your operating system. When issues occur, share screenshots, error messages, and details about what you tried. Working through issues is part of implementing professional projects.

Success

After completing Phase 1. Start & Run, you'll have your own GitHub project running with Kafka.

Use four named terminals:

1. kafka - keep the Kafka message broker running
2. topics - create, list, or reset Kafka topics
3. producer - run the project and producer
4. consumer - run the consumer

After the producer and consumer run successfully, you should see:

========================
Consumer executed successfully!
========================

A new file project.log will appear in the root project folder and processed data will appear in data/output/.

Command Reference

The commands below are used in the workflow guide above. They are provided here for convenience.

Important: the first few times you run a project, follow the guide with the complete instructions.

Show command reference

In a machine terminal (open in your Repos folder)

After you get a copy of this repo in your own GitHub account, open a machine terminal in your Repos folder:

# Replace username with YOUR GitHub username.
git clone https://github.com/bjdawson23/streaming-03-analytics

cd streaming-03-analytics
code .

In VS Code Terminal 1: Start Kafka (kafka)

For full instructions see start kafka.

If any command fails, repeat the steps at install kafka until starting up is reliable.

Open a new VS Code terminal. Rename it kafka. If running Windows, specify the terminal type as wsl or type wsl. Run the commands one at a time.

Step 1. Verify Java and PATH

echo "$JAVA_HOME"

"$JAVA_HOME/bin/java" --version

Step 2. Rebuild ClusterID (as needed)

cd ~/kafka

rm -rf /tmp/kraft-combined-logs

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

echo "Cluster ID: $KAFKA_CLUSTER_ID"

bin/kafka-storage.sh format --standalone -t "$KAFKA_CLUSTER_ID" -c config/server.properties

Step 3. Start kafka server (keep running)

cd ~/kafka

bin/kafka-server-start.sh config/server.properties

In VS Code terminal 2: Create Topic (topics)

For full instructions see create topic.

The topic name must match the name defined in your .env file (copy .env.example to .env).

Open another VS Code terminal. Rename it topics. If running Windows, specify the terminal type as wsl or type wsl. Run the commands one at a time.

cd ~/kafka

bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --partitions 1 \
  --replication-factor 1 \
  --topic streaming-03-analytics-dawson

In VS Code Terminal 3: Run Project and Producer (producer)

Open another VS Code terminal. Rename it producer. If running Windows, use PowerShell. Run the commands one at a time.

# reset uv cache only if/when you start getting strange dependency errors
# uv cache clean

uv self update
uv python pin 3.14
uv sync --extra dev --extra docs --upgrade

uvx pre-commit install

git add -A
uvx pre-commit run --all-files
# repeat if changes were made
git add -A
uvx pre-commit run --all-files

# run the producer
clear
uv run python -m streaming.kafka_producer_case
uv run python -m streaming.kafka_producer_dawson

# do chores
uv run ruff format .
uv run ruff check . --fix
uv run python -m pyright
uv run python -m pytest
uv run python -m zensical build

# save progress
git add -A
git commit -m "update"
git push -u origin main

In VS Code Terminal 4: Run Consumer (consumer)

Open another VS Code terminal. Rename it consumer. If running Windows, use PowerShell. Run the commands one at a time. Clear the terminal, then start the consumer.

clear
uv run python -m streaming.kafka_consumer_dawson

To start fresh, see manage topics to delete the topic and recreate it.

Python Files

src/streaming/data_engineering/derived_fields_dawson.py
src/streaming/data_validation/data_validation_dawson.py
src/streaming/data_validation/data_contract_dawson
src/streaming/kafka_consumer_dawson.py
src/streaming/kafka_producer_dawson.py
src/streaming/kafka_admin_dawson.py

Phase 4 & 5 Changes

Phase 4. Changed KAFKA_CLEAR_TOPIC_ON_START=false in .env.
  That means it will reprocess older messages already stored in the topic and creating duplicates.
Phase 4. Changed KAFKA_CLEAR_TOPIC_ON_START=true in .env (back to original setting)
  This shows the latest consumed messages only.
Phase 4. Changed the Kafka topic set in .env: streaming-03-analytics-dawson
Phase 5. I modified the consumer to write an additional summary CSV and generate a chart that shows payment methods by region.

Results

Describe what happened when you ran the producer and consumer.

The producer ran successfully and delivered the sales messages to Kafka.

The consumer also ran successfully and processed the messages from the topic.

In the most recent run, 178 messages were accepted and 0 were skipped.

The main output CSV contains the accepted sales records with the derived fields subtotal, tax_amount, and total.

The additional output files are payment_methods_by_region_dawson.csv and payment_methods_by_region_dawson.png in data/output/.

The logs show each message being validated, enriched, accepted, and included in the running summary statistics.

More documentation can be found in: docs/ - the project narrative and documentation

Interpretation

Compared with the original example, this version adds project-specific validation, enrichment, and reporting for a sales stream.

I learned that validation is essential in a streaming pipeline because bad records need to be filtered before they affect calculations or reports.

I also learned that enrichment works best when it stays close to the consumer, where reference data is available and calculations can be done immediately as messages arrive.

The running summaries show how revenue changes over time and help reveal typical order sizes as well as outliers.

The payment-method-by-region output adds business intelligence by showing where different payment options are used most often, which could help with marketing, fraud review, or checkout optimization.

Notes

• Use the UP ARROW and DOWN ARROW in the terminal to scroll through past commands.
• Use CTRL+f to find (and replace) text within a file.
• You do not need to add to or modify tests/. They are provided for example only.
• Many files are silent helpers. Explore as you like, but nothing is required.
• You do NOT not to understand everything; understanding builds naturally over time.

Troubleshooting >>> or

If you see something like this in your terminal: >>> or ... You accidentally started Python interactive mode. It happens. Press Ctrl+c (both keys together) or Ctrl+Z then Enter on Windows.