Streaming Data¶
This site provides documentation for this project. Use the navigation to explore module-specific materials.
How-To Guide¶
Many instructions are common to all our projects.
See ⭐ Workflow: Apply Example to get these projects running on your machine.
Project Documentation Pages (docs/)¶
- Home - this documentation landing page
- Project Instructions - instructions specific to this module
- Your Files - how to copy from examples and make them yours
- Glossary - project terms and concepts
- API - autogenerated look at the code interface
Custom Project¶
Dataset¶
The Kafka producer reads sales transactions from data/sales.csv. Each row includes fields such as order_id, datetime, region_id, currency_code, product_id, unit_price, quantity, is_online, customer_id, payment_method, referral_source, discount_code, and customer_note.
This project uses a copy of the original dataset, along with regions.csv and products.csv as reference data for validation and enrichment.
Data Contract¶
Sales messages must include order_id, datetime, region_id, currency_code, product_id, unit_price, quantity, is_online, customer_id, and payment_method.
Optional fields are is_new_customer, device_type, referral_source, discount_code, and customer_note.
The consumer also checks that region_id exists in regions.csv, product_id exists in products.csv, and the text fields match the allowed values defined in the contract. That includes currency_code, payment_method, device_type, and referral_source.
A message is valid when all required fields are present and every field passes the contract rules. Missing fields or disallowed values make the message invalid.
Kafka Messages¶
The producer sends one sales row from data/sales.csv at a time to the Kafka topic streaming-03-analytics-dawson.
Each message uses region_id as the key so related messages stay grouped together in Kafka.
The producer sends only the raw sales fields. Derived fields are added later by the consumer.
Consumer Validation¶
The consumer validates each message before it performs any calculations. It checks required fields, allowed values, reference-table matches for regions and products, and field-level rules for datetime, quantity, and boolean text values.
Valid messages are enriched and written to the consumed output CSV. Invalid messages are skipped and logged.
This validation step protects the downstream summaries from bad source records.
Data Engineering and Enrichment¶
For each accepted message, the consumer computes three derived fields: subtotal, tax_amount, and total. subtotal comes from quantity * unit_price, tax_amount comes from the region tax rate in regions.csv, and total is the sum of the two.
The region tax rate is stored as a percentage in the lookup table, so the consumer converts it to a decimal before calculating tax.
I also added a second output that summarizes message counts by region_id and payment_method, then saves the results as both a CSV and a chart.
Streaming Analytics¶
The consumer tracks running sales statistics from the accepted message totals. As messages arrive, it logs the running total, average, minimum, and maximum sale value.
Over the run, those values change steadily as new orders are accepted, which gives an immediate picture of the stream’s behavior.
The payment-method-by-region summary adds a second view of the stream by showing which payment methods are used most often in each region.
Experiments¶
For Phase 4, I changed the producer message count in .env so the producer could process 175 messages instead of the default 6.
For Phase 5, I extended the consumer to write an additional summary CSV and generate a chart that shows payment methods by region.
That gave me a small but meaningful extension of the original streaming pattern without changing the overall pipeline structure.
Results¶
The producer ran successfully and delivered the sales messages to Kafka. The consumer also ran successfully and processed the topic from start to finish.
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 outputs 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.
Interpretation¶
Compared with the original example, this version adds project-specific validation, enrichment, and reporting for a sales stream.
The biggest lesson was that validation has to happen before analytics. If bad records are allowed through, they can distort both the calculations and the final reports.
I also saw that enrichment is most useful when it happens close to the consumer, where the reference data is already available and the calculations can be applied immediately.
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.