thanhtai435/olist-r2-pipeline

Olist Brazilian E-Commerce → Cloudflare R2 Pipeline

Data pipeline chuyển đổi 2 dataset Olist từ Kaggle lên Cloudflare R2 theo kiến trúc Medallion (Bronze → Silver → Gold), tối ưu cho Data Engineering projects.

Datasets

Dataset	Kaggle	Files	Rows
Brazilian E-Commerce	olistbr/brazilian-ecommerce	9 CSVs	~1.5M
Marketing Funnel	olistbr/marketing-funnel-olist	2 CSVs	~9K

Kiến trúc R2

s3://olist-data-lake/
├── raw/                          # Bronze: CSV nguyên bản (immutable)
│   ├── ecommerce/                # 9 CSV từ brazilian-ecommerce
│   └── marketing_funnel/         # 2 CSV từ marketing-funnel-olist
│
├── staging/                      # Silver: Parquet typed, 1 file/bảng
│   ├── customers.parquet         # ~99K rows
│   ├── orders.parquet            # ~99K rows, timestamps đã cast chuẩn
│   ├── order_items.parquet       # ~113K rows
│   ├── order_payments.parquet    # ~104K rows
│   ├── order_reviews.parquet     # ~99K rows
│   ├── products.parquet          # ~33K rows
│   ├── sellers.parquet           # ~3K rows
│   ├── geolocation.parquet       # DEDUP: 1M→~19K zip codes
│   ├── category_translation.parquet
│   ├── marketing_qualified_leads.parquet
│   └── closed_deals.parquet
│
└── analytics/                    # Gold: Hive-partitioned by year/month
    ├── orders/
    ├── order_items/              # Đã join orders để có timestamp
    ├── seller_performance/       # Seller-level metrics
    └── funnel/                   # Lead-to-deal + days_to_close

ERD - Mối quan hệ giữa các bảng

Marketing Funnel:                     E-Commerce:
  marketing_qualified_leads ──┐         customers ──▶ orders ──▶ order_items ◀── products
         │                     │              ▲                     │
         └── closed_deals ─────┼── seller_id ─┘               order_payments
                               │                              order_reviews
                               └── seller_id ──▶ sellers ◀── geolocation

Link: closed_deals.seller_id = sellers.seller_id → phân tích funnel-to-revenue.

Cài đặt & Chạy

Prerequisites

pip install duckdb boto3 kagglehub

Chạy pipeline

# Cách 1: Truyền credentials trực tiếp
python olist_r2_pipeline.py \
  --r2-endpoint "https://<account_id>.r2.cloudflarestorage.com" \
  --r2-access-key "<your_access_key>" \
  --r2-secret "<your_secret_key>"

# Cách 2: Qua environment variables
export R2_ENDPOINT_URL="https://<account_id>.r2.cloudflarestorage.com"
export R2_ACCESS_KEY_ID="..."
export R2_SECRET_ACCESS_KEY="..."
python olist_r2_pipeline.py

# Test local trước (không upload lên R2)
python olist_r2_pipeline.py --skip-download --skip-upload

Options

Flag	Mô tả
--r2-endpoint	Cloudflare R2 endpoint URL
--r2-access-key	R2 Access Key ID
--r2-secret	R2 Secret Access Key
--skip-download	Bỏ qua download từ KaggleHub
--skip-upload	Chỉ build local, không upload R2
--skip-validate	Bỏ qua bước validate trên R2

Query dữ liệu trên R2 bằng DuckDB

Sau khi upload, query trực tiếp R2 qua DuckDB (zero egress fee):

-- Setup một lần
CREATE SECRET (TYPE r2, KEY_ID '...', SECRET '...', ACCOUNT_ID '...');

-- Query Silver layer (toàn bộ dataset ~30MB Parquet)
SELECT c.customer_state, COUNT(DISTINCT o.order_id) AS orders, SUM(oi.price) AS revenue
FROM read_parquet('r2://olist-data-lake/staging/orders.parquet') o
JOIN read_parquet('r2://olist-data-lake/staging/customers.parquet') c ON o.customer_id = c.customer_id
JOIN read_parquet('r2://olist-data-lake/staging/order_items.parquet') oi ON o.order_id = oi.order_id
WHERE o.order_status = 'delivered'
GROUP BY 1 ORDER BY 2 DESC;

-- Query Gold với partition pruning (chỉ scan tháng cần)
SELECT * FROM read_parquet(
  'r2://olist-data-lake/analytics/orders/**/*.parquet',
  hive_partitioning = true
) WHERE year = '2017' AND month = '06';

-- Funnel analysis (đã join sẵn, có days_to_close)
SELECT origin, COUNT(*) AS leads, AVG(days_to_close) AS avg_days
FROM read_parquet('r2://olist-data-lake/analytics/funnel/**/*.parquet', hive_partitioning=true)
GROUP BY 1;

Các quyết định thiết kế

Quyết định	Giải thích
Parquet thay vì CSV	Columnar + compressed → DuckDB đọc nhanh hơn 10-100x qua network
Snappy compression	Cân bằng tốc độ đọc/ghi + tỉ lệ nén (~3x so với CSV)
Hive partition year/month	Hầu hết query DE filter theo thời gian → partition pruning
geolocation dedup	1M rows → 19K unique zip codes → giảm 50x
TRY_CAST everywhere	Chịu được cả VARCHAR và TIMESTAMP từ read_csv_auto
Funnel pre-joined ở Gold	Có sẵn days_to_close, không cần join lại mỗi lần query
Bảng nhỏ không partition	Sellers (3K), categories (71) giữ single file ở staging

Schema chi tiết

orders (Silver)

Column	Type
order_id	VARCHAR
customer_id	VARCHAR
order_status	VARCHAR
order_purchase_timestamp	TIMESTAMP
order_approved_at	TIMESTAMP
order_delivered_carrier_date	TIMESTAMP
order_delivered_customer_date	TIMESTAMP
order_estimated_delivery_date	TIMESTAMP

funnel mart (Gold) - đã join sẵn

Column	Type	Mô tả
mql_id	VARCHAR	Marketing Qualified Lead ID
seller_id	VARCHAR	FK → sellers
won_date	TIMESTAMP	Ngày chốt deal
business_segment	VARCHAR	Phân khúc
lead_type	VARCHAR	online_medium, online_small...
lead_behaviour_profile	VARCHAR	cat, wolf, eagle, shark
origin	VARCHAR	Nguồn traffic (organic_search, paid_search...)
first_contact_date	TIMESTAMP	Ngày first touch
days_to_close	INTEGER	Số ngày từ MQL → won

Xem đầy đủ schema của tất cả bảng trong source code.

Prerequisites cho R2

1. Tạo R2 bucket tại Cloudflare Dashboard → R2
2. Tạo API token với quyền Object Read & Write
3. Lấy:
   • Account ID (32 ký tự hex)
   • Access Key ID
   • Secret Access Key
   • Endpoint URL: https://<account_id>.r2.cloudflarestorage.com

License

MIT