Assignment: ShopStream

Part 1: Scope and requirements

---

The story

It is Monday morning. You are a new data engineer at ShopStream, an online shop that sells across India.

The big festival sale starts on Friday. Last year, the sales team had a problem. Their numbers came from a job that ran once every night. During the sale, they did not know how much they were selling until the next day. They could not move stock. They could not fix problems in time.

Your manager, Anita, calls you to her desk.

"This year, we use Kafka," she says. "Every order event goes into a topic. I need three things by Friday:

1. Every valid order saved in a database, with its latest status.
2. Live sales numbers for each delivery zone.
3. A way to look back and rebuild the numbers for any period, if something goes wrong."

She also gives you a warning.

"Our partner apps send messy data. Some events are broken. Do not let one bad event stop the pipeline. And do not lose good events, even if a program crashes."

This assignment is that project. You will build it, step by step, using what you learned in the course.

---

What you will build

You will build five small programs.

Here is the big picture.

The big picture of ShopStream

How long will it take? Plan for about 3 to 4 hours. You can split it over two days.

---

What you need before you start

• The 3-broker Kafka cluster from the course, running
• Python 3 and the confluent-kafka library
• The starter kit (a separate download)

The starter kit has:

Get ready like this:

pip install -r requirements.txt
python setup_topics.py

Write all your programs in the same folder as these files.

---

The data

Delivery zones

ShopStream has four delivery zones: NORTH, SOUTH, EAST and WEST.

Topics

setup_topics.py creates both. It also sets min.insync.replicas to 2.

Event types

Every event is a JSON object. There are five types.

An ORDER_CREATED event looks like this:

jsonCopy
{"event_id": "E-CRE-0000", "event_type": "ORDER_CREATED", "order_id": "O-1001",
 "zone": "SOUTH", "customer_id": "C-008", "amount": 4999, "currency": "INR"}

The other four types are shorter. They have no customer and no amount:

jsonCopy
{"event_id": "E-SHI-0076", "event_type": "ORDER_SHIPPED", "order_id": "O-1077", "zone": "SOUTH"}

The fields

The messy truth

The data files are not clean. Look inside them. You will find:

• lines that are not JSON at all
• events with a missing field
• events with a zone that does not exist
• events with an amount that is zero, negative or text
• events with an unknown event type
• the same event sent twice
• status events that arrive in the wrong order (for example, DELIVERED before SHIPPED)

Your pipeline must deal with all of these.

---

Requirements

A. Both producers

The two producers are two separate programs. They share the same rules. You can put the shared code in one module.

B. Order Loader

C. Sales Aggregator

D. Replay tool

---

The database contract

Your checker (verify_results.py) reads your databases. So your tables must use these names.

orders.db, table orders

You may add more columns or tables. These columns must exist.

sales.db, table sales_by_minute

---

How to check your work

Start your consumers. Send the data with both producers. Wait until the consumers have caught up. Stop them with Ctrl+C. Then run:

python verify_results.py

You will see something like this when everything is right:

1. orders.db  (made by the Order Loader)
  [PASS] number of orders
  [PASS] orders and amount per zone
  [PASS] orders per status

2. sales.db  (made by the Sales Aggregator)
  [PASS] orders and revenue per zone

3. Dead-letter topic  (bad messages)
  [PASS] number of messages in the DLQ

ALL CHECKS PASSED. Well done!

If a check fails, the checker shows what it expected and what it found.

Do these failure tests too

Passing the checker is not the end. Prove that your pipeline is strong.

1. Two loaders. Start two Order Loaders. Check that each one gets different partitions.
2. Stop one loader. While data is flowing, press Ctrl+C in one loader. The other must take over its partitions.
3. Crash one loader. Stop a loader with kill -9 (or end the task). Start it again. Run the checker. The numbers must still be right. Note: Kafka may take up to 45 seconds to notice that a member is gone.
4. Stop a broker. Stop one of the three Kafka brokers while the producers run. Nothing should be lost.
5. Replay. Note the time when you start the producers. When everything is done, run the replay tool for that time. It must print MATCH.

---

Rules and tips

• Do not hard-code the expected numbers in your programs. Your programs must work for any valid data.
• To start again from zero, run python setup_topics.py --reset. This deletes the topics and your two database files.
• Start the consumers before the producers if you want to watch the work live.
• If something does not work, read the error first. Then check the container logs.
• Try the assignment on your own first. The next two parts help you: Part 2 explains the design, Part 3 walks you through a full solution.

---

Extra challenges (optional)

When you finish, try one of these on your own:

1. Add a --zone option to the Sales Aggregator, so one copy reads only that zone. Hint: use assign() on the zone's partition.
2. Make the loader save in batches of 20 events instead of one by one. What changes in your safe order?
3. Add a third consumer group that writes every ORDER_CANCELLED event to a text file.