Library Project - Batch ETL va Realtime CDC Pipeline

Project nay minh hoa mot he thong data engineering gom 2 luong doc lap:

Batch ETL: MySQL -> Spark -> MinIO theo Bronze/Silver/Gold.
Realtime CDC: MySQL binlog -> Debezium -> Kafka -> Python consumers -> Redis -> Kafka order_final.

Batch dung de tao data warehouse. Realtime CDC dung de bat order moi gan realtime.

Mental Model

Ngan gon:

MySQL = source database
Airflow = scheduler crawl books
Debezium = doc MySQL binlog
Kafka = event bus
Python consumers = xu ly realtime logic
Redis = state tam + dedup
Spark = batch transform engine
MinIO = data lake luu Parquet
Bronze = raw
Silver = cleaned
Gold = analytics/star schema

Batch flow:

MySQL -> Spark Bronze -> Spark Silver -> Spark Gold -> MinIO

Realtime flow:

MySQL binlog -> Debezium -> Kafka CDC topics -> Python consumers -> Redis -> Kafka order_final

Batch va realtime khong can chay cung luc.

Neu muon CDC bat dung order moi:

Start Debezium + consumers truoc -> generate/insert orders sau

Thanh Phan He Thong

Component	Container	Port	Vai tro
MySQL	`mysql-db`	`3306`	Source database
Zookeeper	`zookeeper`	`2181`	Kafka coordination
Kafka broker 1	`kafka-broker`	`9092`, `9093`	Kafka broker
Kafka broker 2	`kafka-broker-2`	`9094`, `9095`	Kafka broker
Kafka Connect	`kafka-connect`	`8083`	Chay Debezium connector
Redis	`redis-cache`	`6379`	Cache + dedup realtime
MinIO	`minio-storage`	`9000`, `9001`	S3-compatible data lake
Spark master	`spark-master`	`7077`, `8080`	Spark cluster master
Spark worker	`spark-worker`	`8081`	Spark executor worker
Airflow webserver	`airflow-webserver`	`8082`	Airflow UI
Airflow scheduler	`airflow-scheduler`	-	Schedule DAG
Kafka UI	`kafka-ui`	`8084`	Xem Kafka topics

Kien Truc Tong Quan

                       +------------------+
                       |      MySQL       |
                       | project_db       |
                       | books            |
                       | customers        |
                       | payment_method   |
                       | orders           |
                       | order_details    |
                       +--------+---------+
                                |
             +------------------+------------------+
             |                                     |
             v                                     v
   +---------------------+              +---------------------+
   | Batch ETL           |              | Realtime CDC        |
   | Spark JDBC          |              | Debezium            |
   +----------+----------+              +----------+----------+
              |                                    |
              v                                    v
   +---------------------+              +---------------------+
   | MinIO               |              | Kafka topics        |
   | bronze              |              | mysql.project_db.*  |
   | silver              |              +----------+----------+
   | gold                |                         |
   +---------------------+                         v
                                      +------------------------+
                                      | Python consumers       |
                                      | order.py               |
                                      | order_details.py       |
                                      +-----------+------------+
                                                  |
                                                  v
                                      +------------------------+
                                      | Redis                  |
                                      | order_info:*           |
                                      | ordered_products:*     |
                                      | order_status:*         |
                                      +-----------+------------+
                                                  |
                                                  v
                                      +------------------------+
                                      | Kafka order_final      |
                                      +------------------------+

Cau Truc Thu Muc

.
├── docker-compose.yml
├── requirements.txt
├── .env.example
├── data/
│   ├── customers.csv
│   └── payment_method.csv
├── init-scripts/
│   ├── 01_init.sql
│   └── run.md
├── airflow/
│   └── dags/
│       └── openlibrary_dag.py
├── scripts/
│   ├── crawl-data/
│   │   ├── crawl_openlibrary.py
│   │   ├── db_helper.py
│   │   └── checkpoint.py
│   ├── database/
│   │   ├── load_file.py
│   │   └── generate_orders.py
│   ├── batch/
│   │   ├── bronze_layer.py
│   │   ├── silver_layer.py
│   │   └── gold_layer.py
│   ├── real-time/
│   │   ├── mysql-src-connector.json
│   │   ├── kafka_handler.py
│   │   ├── order.py
│   │   └── order_details.py
│   └── utils.py
├── tests/
│   └── test_regressions.py
├── logs/
└── checkpoints/

Cach Chay Tu Dau

Tat ca command ben duoi chay trong PowerShell tai root project:

cd D:\Data\Library_project

1. Cai Python dependencies

python -m pip install -r requirements.txt

2. Start Docker stack

docker compose up -d

Check:

docker compose ps

3. Check Kafka Connect

curl.exe http://localhost:8083/connectors

Neu chua co connector:

curl.exe -X POST http://localhost:8083/connectors -H "Content-Type: application/json" --data-binary "@scripts/real-time/mysql-src-connector.json"

Check status:

curl.exe http://localhost:8083/connectors/mysql-source-connector/status

Phai thay:

connector RUNNING

4. Cap quyen MySQL cho Debezium

Chay lenh nay neu connector bi loi privilege:

docker exec mysql-db mysql -uroot -padmin -e "GRANT RELOAD, FLUSH_TABLES, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'admin'@'%'; FLUSH PRIVILEGES;"

Restart connector:

curl.exe -X POST http://localhost:8083/connectors/mysql-source-connector/restart

5. Load customers va payment methods

$env:MYSQL_HOST='localhost'
python scripts/database/load_file.py

6. Crawl/load books

Airflow DAG:

docker exec -it airflow-webserver airflow dags trigger openlibrary_crawler

Quan trong:

Neu openlibrary.org bi timeout, khong co data sach that.
Project hien tai mac dinh khong force offline de tranh ghi Offline Book ... gia.
Neu muon test offline, set OPENLIBRARY_FORCE_OFFLINE=true trong .env roi restart Airflow.

Test OpenLibrary tu container:

docker exec airflow-webserver python -c "import urllib.request; print(urllib.request.urlopen('https://openlibrary.org/search.json?q=fiction&limit=1&offset=0', timeout=20).status)"

Neu tra 200, co the crawl data that.

7. Generate orders

Chi generate sau khi books, customers, payment_method da co data.

$env:MYSQL_HOST='localhost'
$env:MAX_ORDERS='1000'
$env:ORDER_SLEEP_SECONDS='0'
python scripts/database/generate_orders.py

Check MySQL:

docker exec mysql-db mysql -uadmin -padmin project_db -e "SELECT COUNT(*) AS books FROM books; SELECT COUNT(*) AS customers FROM customers; SELECT COUNT(*) AS payment_method FROM payment_method; SELECT COUNT(*) AS orders FROM orders; SELECT COUNT(*) AS order_details FROM order_details;"

8. Run batch ETL

Khong chay batch bang python scripts/batch/*.py tren Windows.

Chay trong Spark container:

docker exec spark-master /opt/spark/bin/spark-submit --master spark://spark-master:7077 --packages org.apache.hadoop:hadoop-aws:3.3.4,mysql:mysql-connector-java:8.0.33 /opt/scripts/batch/bronze_layer.py

docker exec spark-master /opt/spark/bin/spark-submit --master spark://spark-master:7077 --packages org.apache.hadoop:hadoop-aws:3.3.4,mysql:mysql-connector-java:8.0.33 /opt/scripts/batch/silver_layer.py

docker exec spark-master /opt/spark/bin/spark-submit --master spark://spark-master:7077 --packages org.apache.hadoop:hadoop-aws:3.3.4,mysql:mysql-connector-java:8.0.33 /opt/scripts/batch/gold_layer.py

9. Run realtime consumers

Mo terminal PowerShell 1:

$env:KAFKA_BOOTSTRAP='localhost:9093,localhost:9095'
$env:REDIS_HOST='localhost'
python scripts/real-time/order.py

Mo terminal PowerShell 2:

$env:KAFKA_BOOTSTRAP='localhost:9093,localhost:9095'
$env:REDIS_HOST='localhost'
python scripts/real-time/order_details.py

Neu dung CMD, dung set, khong dung $env:

set KAFKA_BOOTSTRAP=localhost:9093,localhost:9095
set REDIS_HOST=localhost
python scripts/real-time/order_details.py

10. Test realtime bang order moi

Sau khi Debezium va consumers da chay, insert order moi:

docker exec mysql-db mysql -uadmin -padmin project_db -e "SET @oid=UUID(); INSERT INTO orders (id, timestamp, customer_id, store_id, payment_method_id, num_products) VALUES (@oid, NOW(), 1, 1, 1, 2); INSERT INTO order_details (order_id, product_id, quantity, subtotal, is_suggestion) VALUES (@oid, 1, 1, 13.82, false), (@oid, 2, 2, 27.66, false); SELECT @oid AS test_order_id;"

Check output:

docker exec kafka-broker kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic order_final --time -1

Xem messages:

docker exec kafka-broker kafka-console-consumer --bootstrap-server localhost:9092 --topic order_final --from-beginning --max-messages 5

Batch ETL Pipeline

Batch ETL xu ly du lieu theo lo. No doc MySQL tai mot thoi diem va ghi data lake tren MinIO.

Source tables

books
customers
payment_method
orders
order_details

Bronze Layer

File:

scripts/batch/bronze_layer.py

Input:

MySQL

Output:

s3a://bronze/orders
s3a://bronze/order_details
s3a://bronze/customers
s3a://bronze/payment_method
s3a://bronze/books

Vai tro:

Copy raw data tu MySQL sang MinIO.
Them metadata:

bronze_ingestion_time
bronze_source
month

Silver Layer

File:

scripts/batch/silver_layer.py

Input:

s3a://bronze/...

Output:

s3a://silver/orders
s3a://silver/order_details
s3a://silver/customers
s3a://silver/payment_method

Vai tro:

Dedupe.
Clean schema.
Rename ID cho ro nghia.
Giu quantity va subtotal that.

Schema quan trong:

orders.id -> orders.order_id
customers.id -> customers.customer_id
payment_method.id -> payment_method.payment_method_id
order_details.id -> order_details.order_detail_id

Gold Layer

File:

scripts/batch/gold_layer.py

Output dimensions:

s3a://gold/dimensions/dim_customer
s3a://gold/dimensions/dim_payment_method
s3a://gold/dimensions/dim_product
s3a://gold/dimensions/dim_date

Output facts:

s3a://gold/facts/fact_orders
s3a://gold/facts/fact_order_products

fact_orders gom theo order:

order_id
customer_id
payment_method_id
order_date_id
product_count
expected_products
quantity_ordered
order_status
created_timestamp
fact_created_at

total_amount tinh tu:

sum(order_details.subtotal)

quantity_ordered tinh tu:

sum(order_details.quantity)

Realtime CDC Pipeline

Realtime CDC bat thay doi moi tu MySQL.

MySQL binlog -> Debezium -> Kafka -> Python consumers -> Redis -> Kafka order_final

Can chay cung luc cai gi?

De bat CDC dung, can chay:

MySQL
Kafka
Kafka Connect / Debezium connector
Redis
order.py
order_details.py

Batch khong can chay cung realtime.

Thu tu dung:

Start Debezium + consumers truoc -> generate/insert orders sau

Debezium lam gi?

Debezium doc MySQL binlog.

Connector:

mysql-source-connector

Config:

scripts/real-time/mysql-src-connector.json

No doc 2 tables:

project_db.orders
project_db.order_details

Va ghi ra Kafka topics:

mysql.project_db.orders
mysql.project_db.order_details

Debezium message

Message co dang:

{
  "payload": {
    "before": null,
    "after": {
      "id": "order-id",
      "customer_id": 1,
      "payment_method_id": 1,
      "num_products": 2
    },
    "op": "c"
  }
}

op:

c = create/insert
u = update
r = snapshot/read

`order.py`

Consumes:

mysql.project_db.orders

Consumer group:

order_info_tracker

Vai tro:

Doc order metadata.
Luu Redis hash:

order_info:{order_id}

Gia tri:

customer_id
payment_method_id
num_products

order.py khong emit order_final.

`order_details.py`

Consumes:

mysql.project_db.order_details

Consumer group:

order_products_tracker

Vai tro:

Doc product trong order.
Add product vao Redis set:

ordered_products:{order_id}

Sau moi product, check:

len(ordered_products:{order_id}) == order_info:{order_id}.num_products

Neu bang nhau, order complete.

Redis keys

order_info:{order_id}
ordered_products:{order_id}
order_status:{order_id}

order_status:{order_id} dung de dedup:

emitting
completed

Flow dedup:

SETNX order_status:{order_id} emitting
send order_final
set order_status:{order_id} completed

Neu worker khac thay key da ton tai, no khong emit nua.

`order_final`

Topic output:

order_final

Message mau:

{
  "order_id": "69339232-7905-11f1-a1f4-7a7e11b212fe",
  "customer_id": "1",
  "payment_method_id": "1",
  "product_ids": ["1", "2"]
}

No nghia la:

Order nay da du order metadata va du product details.

DBeaver Va UI

DBeaver - MySQL

Config:

Database: MySQL
Host: localhost
Port: 3306
Database: project_db
Username: admin
Password: admin

Tables thay trong DBeaver:

books
customers
payment_method
orders
order_details

SQL hay dung:

SHOW TABLES;
SELECT COUNT(*) FROM books;
SELECT COUNT(*) FROM customers;
SELECT COUNT(*) FROM payment_method;
SELECT COUNT(*) FROM orders;
SELECT COUNT(*) FROM order_details;

Luu y:

Bronze/Silver/Gold khong nam trong MySQL.
Chung la Parquet files tren MinIO.
DBeaver MySQL khong thay bronze, silver, gold.

MinIO UI

URL: http://localhost:9001
Username: admin
Password: admin123

Buckets:

bronze
silver
lookup-data
order-info

Airflow UI

URL: http://localhost:8082
Username: admin
Password: admin

DAG:

openlibrary_crawler

Kafka UI

URL: http://localhost:8084

Spark UI

URL: http://localhost:8080

Kiem Tra Va Debug

Check services

docker compose ps

Check connector

curl.exe http://localhost:8083/connectors/mysql-source-connector/status

Check Kafka topics

docker exec kafka-broker kafka-topics --bootstrap-server localhost:9092 --list

Check Kafka offsets

docker exec kafka-broker kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic mysql.project_db.orders --time -1
docker exec kafka-broker kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic mysql.project_db.order_details --time -1
docker exec kafka-broker kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic order_final --time -1

Consume `order_final`

docker exec kafka-broker kafka-console-consumer --bootstrap-server localhost:9092 --topic order_final --from-beginning --max-messages 5

Check Redis

docker exec redis-cache redis-cli -n 1 KEYS "order_info:*"
docker exec redis-cache redis-cli -n 1 KEYS "ordered_products:*"
docker exec redis-cache redis-cli -n 1 KEYS "order_status:*"

Check MySQL counts

docker exec mysql-db mysql -uadmin -padmin project_db -e "SELECT COUNT(*) AS books FROM books; SELECT COUNT(*) AS customers FROM customers; SELECT COUNT(*) AS payment_method FROM payment_method; SELECT COUNT(*) AS orders FROM orders; SELECT COUNT(*) AS order_details FROM order_details;"

Run tests

python -m pytest tests

Reset Du Lieu

Reset MySQL transactional tables

Lenh nay xoa orders, order_details, books.

docker exec mysql-db mysql -uadmin -padmin project_db -e "SET FOREIGN_KEY_CHECKS=0; TRUNCATE TABLE order_details; TRUNCATE TABLE orders; TRUNCATE TABLE books; SET FOREIGN_KEY_CHECKS=1;"

Xoa checkpoint crawler:

Remove-Item -LiteralPath "checkpoints\openlibrary_checkpoint.txt" -Force -ErrorAction SilentlyContinue

Reset Redis realtime state

docker exec redis-cache redis-cli -n 1 FLUSHDB

Reset Kafka realtime topics

Can can than. Lenh nay xoa topics.

docker exec kafka-broker kafka-topics --bootstrap-server localhost:9092 --delete --topic mysql.project_db.orders
docker exec kafka-broker kafka-topics --bootstrap-server localhost:9092 --delete --topic mysql.project_db.order_details
docker exec kafka-broker kafka-topics --bootstrap-server localhost:9092 --delete --topic order_final

Recreate connector:

curl.exe -X DELETE http://localhost:8083/connectors/mysql-source-connector
curl.exe -X POST http://localhost:8083/connectors -H "Content-Type: application/json" --data-binary "@scripts/real-time/mysql-src-connector.json"

Loi Thuong Gap

1. Chay batch bang Python local bi loi Java/Hadoop

Sai:

python scripts/batch/bronze_layer.py

Loi thuong gap:

UnsupportedOperationException: getSubject
winutils.exe not found
HADOOP_HOME unset

Dung:

docker exec spark-master /opt/spark/bin/spark-submit --master spark://spark-master:7077 --packages org.apache.hadoop:hadoop-aws:3.3.4,mysql:mysql-connector-java:8.0.33 /opt/scripts/batch/bronze_layer.py

2. PowerShell multiline dung sai ky tu

PowerShell khong dung \ de xuong dong.

Sai:

docker exec spark-master /opt/spark/bin/spark-submit \
  --master spark://spark-master:7077 \
  /opt/scripts/batch/bronze_layer.py

Dung:

docker exec spark-master /opt/spark/bin/spark-submit `
  --master spark://spark-master:7077 `
  --packages org.apache.hadoop:hadoop-aws:3.3.4,mysql:mysql-connector-java:8.0.33 `
  /opt/scripts/batch/bronze_layer.py

Hoac 1 dong.

3. Ivy cache loi khi dung `--packages`

Loi:

/home/spark/.ivy2/cache/... No such file or directory

Fix:

docker exec -u root spark-master sh -c "mkdir -p /home/spark/.ivy2/cache /home/spark/.ivy2/jars && chown -R spark:spark /home/spark/.ivy2"

4. S3A class not found

Loi:

Class org.apache.hadoop.fs.s3a.S3AFileSystem not found

Nguyen nhan: thieu hadoop-aws jar.

Fix: chay Spark voi --packages:

--packages org.apache.hadoop:hadoop-aws:3.3.4,mysql:mysql-connector-java:8.0.33

5. Debezium connector RUNNING nhung offsets bang 0

Check logs:

docker logs kafka-connect --tail 200

Neu thay:

Access denied; you need RELOAD or FLUSH_TABLES privilege

Fix:

docker exec mysql-db mysql -uroot -padmin -e "GRANT RELOAD, FLUSH_TABLES, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'admin'@'%'; FLUSH PRIVILEGES;"
curl.exe -X POST http://localhost:8083/connectors/mysql-source-connector/restart

6. `order_final` it hon `orders`

Realtime CDC phu hop order moi. Neu consumer start sau snapshot, Redis state co the khong reconstruct hoan hao historical orders.

Muon test sach:

Start Debezium + order.py + order_details.py truoc
Generate/insert orders sau

7. Mo terminal CMD nhung dung lenh PowerShell

PowerShell:

$env:KAFKA_BOOTSTRAP='localhost:9093,localhost:9095'

CMD:

set KAFKA_BOOTSTRAP=localhost:9093,localhost:9095

8. OpenLibrary ra timeout

Test:

docker exec airflow-webserver python -c "import urllib.request; print(urllib.request.urlopen('https://openlibrary.org/search.json?q=fiction&limit=1&offset=0', timeout=20).status)"

Neu timeout, Docker/container khong ra duoc openlibrary.org:443. Can sua proxy/VPN/DNS Docker.

Neu bat OPENLIBRARY_FORCE_OFFLINE=true, data se co dang:

Offline Book 1
Offline Author 1

Neu muon data that, phai de:

OPENLIBRARY_FORCE_OFFLINE=false
OPENLIBRARY_OFFLINE_FALLBACK=false

Trang Thai Da Kiem Chung

Da verify trong qua trinh debug:

Silver Layer completed successfully
Gold Layer completed successfully
Kafka Connect connector RUNNING
Debezium privileges fixed
Kafka topics created
order_final has messages
pytest tests pass

Vi du offsets sau khi chay realtime:

mysql.project_db.orders:0:1001
mysql.project_db.order_details:0:1827
order_final:0:874

Ghi Chu Quan Trong

Batch va realtime la 2 pipeline rieng.
Batch khong bat CDC.
Realtime khong tao Gold warehouse.
DBeaver chi xem MySQL source tables.
MinIO chua Bronze/Silver/Gold Parquet.
Muon CDC bat dung order moi: start Debezium + consumers truoc khi generate orders.
Muon data sach that: Docker/host phai connect duoc OpenLibrary.

Name		Name	Last commit message	Last commit date
Latest commit History 8 Commits
airflow		airflow
checkpoints		checkpoints
data		data
init-scripts		init-scripts
jars		jars
logs		logs
scripts		scripts
tests		tests
.gitignore		.gitignore
README.md		README.md
docker-compose.yml		docker-compose.yml
requirements.txt		requirements.txt

Folders and files

Latest commit

History

Repository files navigation

Library Project - Batch ETL va Realtime CDC Pipeline

Muc Luc

Mental Model

Thanh Phan He Thong

Kien Truc Tong Quan

Cau Truc Thu Muc

Cach Chay Tu Dau

1. Cai Python dependencies

2. Start Docker stack

3. Check Kafka Connect

4. Cap quyen MySQL cho Debezium

5. Load customers va payment methods

6. Crawl/load books

7. Generate orders

8. Run batch ETL

9. Run realtime consumers

10. Test realtime bang order moi

Batch ETL Pipeline

Source tables

Bronze Layer

Silver Layer

Gold Layer

Realtime CDC Pipeline

Can chay cung luc cai gi?

Debezium lam gi?

Debezium message

order.py

order_details.py

Redis keys

order_final

DBeaver Va UI

DBeaver - MySQL

MinIO UI

Airflow UI

Kafka UI

Spark UI

Kiem Tra Va Debug

Check services

Check connector

Check Kafka topics

Check Kafka offsets

Consume order_final

Check Redis

Check MySQL counts

Run tests

Reset Du Lieu

Reset MySQL transactional tables

Reset Redis realtime state

Reset Kafka realtime topics

Loi Thuong Gap

1. Chay batch bang Python local bi loi Java/Hadoop

2. PowerShell multiline dung sai ky tu

3. Ivy cache loi khi dung --packages

4. S3A class not found

5. Debezium connector RUNNING nhung offsets bang 0

6. order_final it hon orders

7. Mo terminal CMD nhung dung lenh PowerShell

8. OpenLibrary ra timeout

Trang Thai Da Kiem Chung

Ghi Chu Quan Trong

About

Resources

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

`order.py`

`order_details.py`

`order_final`

Consume `order_final`

3. Ivy cache loi khi dung `--packages`

6. `order_final` it hon `orders`

Packages