ETL ও ELT: নির্ভরযোগ্য ডেটা পাইপলাইন বানানো

ETL:  উৎস ──extract──► [ Python / Spark-এ transform ] ──load──► ওয়্যারহাউস (শুধু পরিষ্কার টেবিল)

ELT:  উৎস ──extract──► ওয়্যারহাউস (কাঁচা টেবিল) ──ওয়্যারহাউসের ভেতরে SQL দিয়ে transform──► পরিষ্কার টেবিল

import json

deliveries = [
    {"order_id": 1,  "courier": "Pathao",    "delivered_on": "2026-01-14"},
    {"order_id": 2,  "courier": "RedX",      "delivered_on": "2026-01-23"},
    {"order_id": 3,  "courier": "Pathao",    "delivered_on": "2026-02-05"},
    {"order_id": 4,  "courier": "Steadfast", "delivered_on": "2026-02-16"},
    {"order_id": 6,  "courier": "RedX",      "delivered_on": "2026-03-11"},
    {"order_id": 7,  "courier": "Pathao",    "delivered_on": "2026-03-17"},
    {"order_id": 8,  "courier": "Steadfast", "delivered_on": "2026-03-30"},
    {"order_id": 9,  "courier": "RedX",      "delivered_on": "2026-04-06"},
    {"order_id": 10, "courier": "Pathao",    "delivered_on": "2026-04-21"},
    {"order_id": 11, "courier": "Steadfast", "delivered_on": "2026-05-08"},
    {"order_id": 12, "courier": "RedX",      "delivered_on": None},
    {"order_id": 14, "courier": "Pathao",    "delivered_on": "2026-06-12"},
]
with open("courier_api.json", "w", encoding="utf-8") as f:
    json.dump({"deliveries": deliveries}, f, indent=1)

with open("targets.csv", "w", encoding="utf-8") as f:
    f.write("month,target\n2026-01,8000\n2026-02,8000\n2026-03,20000\n"
            "2026-04,20000\n2026-05,15000\n2026-06,20000\n")
print("sources ready")

sources ready

import csv
import logging
import sqlite3
import sys
import time
from datetime import date, timedelta

log = logging.getLogger("etl")

WAREHOUSE_DDL = """
CREATE TABLE IF NOT EXISTS fact_order_lines (
    order_id      INTEGER NOT NULL,
    product_id    INTEGER NOT NULL,
    order_date    TEXT    NOT NULL,
    month         TEXT    NOT NULL,
    category      TEXT    NOT NULL,
    status        TEXT    NOT NULL,
    quantity      INTEGER NOT NULL,
    revenue       INTEGER NOT NULL,
    courier       TEXT,
    delivery_days INTEGER,
    PRIMARY KEY (order_id, product_id)
);
CREATE TABLE IF NOT EXISTS monthly_sales (
    month TEXT PRIMARY KEY, orders INTEGER NOT NULL, revenue INTEGER NOT NULL,
    target INTEGER, pct_of_target REAL
);
CREATE TABLE IF NOT EXISTS etl_state (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS etl_runs (
    run_id INTEGER PRIMARY KEY, mode TEXT, since TEXT, extracted INTEGER,
    inserted INTEGER, updated INTEGER, status TEXT
);
"""

def extract_order_lines(src, since):
    """`since` (YYYY-MM-DD) বা তার পরে দেওয়া অর্ডারের লাইনগুলো।"""
    cur = src.execute("""
        SELECT oi.order_id, oi.product_id, o.order_date, o.status,
               COALESCE(c.name, 'Uncategorised') AS category,
               oi.quantity, oi.quantity * oi.unit_price AS revenue
        FROM order_items oi
        JOIN orders o          ON o.order_id = oi.order_id
        JOIN products p        ON p.product_id = oi.product_id
        LEFT JOIN categories c ON c.category_id = p.category_id
        WHERE o.order_date >= ?
        ORDER BY oi.order_id, oi.product_id""", (since,))
    cols = [d[0] for d in cur.description]
    return [dict(zip(cols, row)) for row in cur]


class CourierAPI:
    """একটা HTTP API-র বদলি। আসল কোডে: requests.get(url, timeout=10).json()।"""
    def __init__(self, path, fail_first=0):
        self.path, self.fail_first = path, fail_first

    def get_deliveries(self):
        if self.fail_first > 0:
            self.fail_first -= 1
            raise ConnectionError("courier API timed out")
        with open(self.path, encoding="utf-8") as f:
            return json.load(f)["deliveries"]


def with_retries(fn, attempts=4, base_delay=0.01):
    """fn() কল করে; সাময়িক এররে base_delay-এর 1, 2, 4 ... গুণ অপেক্ষা করে আবার চেষ্টা করে।"""
    for attempt in range(1, attempts + 1):
        try:
            return fn()
        except ConnectionError as exc:           # সাময়িক এরর: রিট্রাই করার মতো
            if attempt == attempts:
                raise
            delay = base_delay * 2 ** (attempt - 1)
            log.warning("attempt %d failed (%s), retrying in %.2fs", attempt, exc, delay)
            time.sleep(delay)


def extract_targets(path):
    with open(path, newline="", encoding="utf-8") as f:
        return [{"month": r["month"], "target": int(r["target"])} for r in csv.DictReader(f)]

def transform(lines, deliveries):
    by_order = {d["order_id"]: d for d in deliveries}
    rows = []
    for line in lines:
        d = by_order.get(line["order_id"], {})
        days = None
        if d.get("delivered_on"):
            days = (date.fromisoformat(d["delivered_on"]) - date.fromisoformat(line["order_date"])).days
        rows.append({**line, "month": line["order_date"][:7],
                     "courier": d.get("courier"), "delivery_days": days})
    return rows


def validate(rows):
    problems = []
    for r in rows:
        if r["quantity"] <= 0 or r["revenue"] <= 0:
            problems.append(f"order {r['order_id']}: non-positive quantity or revenue")
        if r["delivery_days"] is not None and not 0 <= r["delivery_days"] <= 30:
            problems.append(f"order {r['order_id']}: delivery took {r['delivery_days']} days")
    if problems:
        raise ValueError("validation failed: " + "; ".join(problems))

UPSERT_FACT = """
INSERT INTO fact_order_lines (order_id, product_id, order_date, month, category, status,
                              quantity, revenue, courier, delivery_days)
VALUES (:order_id, :product_id, :order_date, :month, :category, :status,
        :quantity, :revenue, :courier, :delivery_days)
ON CONFLICT (order_id, product_id) DO UPDATE SET
    order_date = excluded.order_date, month = excluded.month, category = excluded.category,
    status = excluded.status, quantity = excluded.quantity, revenue = excluded.revenue,
    courier = excluded.courier, delivery_days = excluded.delivery_days
WHERE (fact_order_lines.status, fact_order_lines.quantity, fact_order_lines.revenue,
       fact_order_lines.courier, fact_order_lines.delivery_days)
   IS NOT (excluded.status, excluded.quantity, excluded.revenue,
           excluded.courier, excluded.delivery_days)
"""

BUILD_MONTHLY = """
INSERT INTO monthly_sales (month, orders, revenue, target, pct_of_target)
SELECT f.month, COUNT(DISTINCT f.order_id), SUM(f.revenue), t.target,
       ROUND(100.0 * SUM(f.revenue) / t.target, 1)
FROM fact_order_lines f LEFT JOIN stg_targets t ON t.month = f.month
WHERE f.status <> 'cancelled'
GROUP BY f.month
ON CONFLICT (month) DO UPDATE SET
    orders = excluded.orders, revenue = excluded.revenue,
    target = excluded.target, pct_of_target = excluded.pct_of_target
WHERE (monthly_sales.orders, monthly_sales.revenue, monthly_sales.target)
   IS NOT (excluded.orders, excluded.revenue, excluded.target)
"""

def count(con, table):
    return con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]

def load(wh, rows, targets):
    """fact সারিগুলো upsert করে, তারপর SQL দিয়ে মাসিক সারাংশ নতুন করে বানায় (ELT ধাপ)।"""
    before, changes = count(wh, "fact_order_lines"), wh.total_changes
    wh.executemany(UPSERT_FACT, rows)
    inserted = count(wh, "fact_order_lines") - before
    updated = wh.total_changes - changes - inserted

    wh.execute("CREATE TEMP TABLE IF NOT EXISTS stg_targets (month TEXT PRIMARY KEY, target INTEGER)")
    wh.execute("DELETE FROM stg_targets")
    wh.executemany("INSERT INTO stg_targets VALUES (:month, :target)", targets)
    changes = wh.total_changes
    wh.execute(BUILD_MONTHLY)
    return inserted, updated, wh.total_changes - changes

def reconcile(src, wh):
    source = src.execute("SELECT COUNT(*), SUM(quantity * unit_price) FROM order_items").fetchone()
    target = wh.execute("SELECT COUNT(*), SUM(revenue) FROM fact_order_lines").fetchone()
    if source != target:
        raise RuntimeError(f"reconciliation failed: source {source} vs warehouse {target}")
    return target

LOOKBACK_DAYS = 14

def run_pipeline(api, src_path="shop.db", wh_path="warehouse.db", full=False):
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s",
                        stream=sys.stdout, force=True)
    src, wh = sqlite3.connect(src_path), sqlite3.connect(wh_path)
    mode = None
    try:
        wh.executescript(WAREHOUSE_DDL)
        state = wh.execute("SELECT value FROM etl_state WHERE key = 'orders_watermark'").fetchone()
        mode = "full" if full or state is None else "incremental"
        since = "0001-01-01" if mode == "full" else \
            (date.fromisoformat(state[0]) - timedelta(days=LOOKBACK_DAYS)).isoformat()

        lines = extract_order_lines(src, since)
        deliveries = with_retries(api.get_deliveries)
        targets = extract_targets("targets.csv")
        log.info("%s extract since %s: %d order lines, %d deliveries, %d targets",
                 mode, since, len(lines), len(deliveries), len(targets))

        rows = transform(lines, deliveries)
        validate(rows)

        with wh:                                   # একটা ট্রানজ্যাকশন: হয় সবটা, নয় কিছুই না
            inserted, updated, monthly = load(wh, rows, targets)
            if rows:
                wh.execute("INSERT INTO etl_state VALUES ('orders_watermark', ?) "
                           "ON CONFLICT (key) DO UPDATE SET value = excluded.value",
                           (max(r["order_date"] for r in rows),))
            wh.execute("INSERT INTO etl_runs (mode, since, extracted, inserted, updated, status) "
                       "VALUES (?, ?, ?, ?, ?, 'ok')", (mode, since, len(rows), inserted, updated))
        n, revenue = reconcile(src, wh)
        log.info("load: %d inserted, %d updated, %d monthly rows changed; warehouse %d lines, %d taka",
                 inserted, updated, monthly, n, revenue)
        return inserted, updated, monthly
    except Exception:
        log.exception("run failed")
        with wh:
            wh.execute("INSERT INTO etl_runs (mode, status) VALUES (?, 'failed')", (mode,))
        raise
    finally:
        src.close()
        wh.close()

run_pipeline(CourierAPI("courier_api.json", fail_first=2))

WARNING etl: attempt 1 failed (courier API timed out), retrying in 0.01s
WARNING etl: attempt 2 failed (courier API timed out), retrying in 0.02s
INFO etl: full extract since 0001-01-01: 20 order lines, 12 deliveries, 6 targets
INFO etl: load: 20 inserted, 0 updated, 6 monthly rows changed; warehouse 20 lines, 103550 taka

import hashlib

def fingerprint(path="warehouse.db"):
    con = sqlite3.connect(path)
    rows = con.execute("SELECT * FROM fact_order_lines ORDER BY order_id, product_id").fetchall()
    rows += con.execute("SELECT * FROM monthly_sales ORDER BY month").fetchall()
    con.close()
    return hashlib.sha256(repr(rows).encode()).hexdigest()[:12]

before = fingerprint()
run_pipeline(CourierAPI("courier_api.json"))
print("unchanged:", fingerprint() == before)

INFO etl: incremental extract since 2026-05-26: 3 order lines, 12 deliveries, 6 targets
INFO etl: load: 0 inserted, 0 updated, 0 monthly rows changed; warehouse 20 lines, 103550 taka
unchanged: True

wh = sqlite3.connect("warehouse.db")
for row in wh.execute("SELECT * FROM monthly_sales ORDER BY month"):
    print(row)
print(wh.execute("""SELECT courier, COUNT(DISTINCT order_id), ROUND(AVG(delivery_days), 1)
                    FROM fact_order_lines WHERE courier IS NOT NULL
                    GROUP BY courier ORDER BY courier""").fetchall())
wh.close()

('2026-01', 2, 6600, 8000, 82.5)
('2026-02', 2, 7000, 8000, 87.5)
('2026-03', 3, 21400, 20000, 107.0)
('2026-04', 2, 23450, 20000, 117.3)
('2026-05', 2, 8500, 15000, 56.7)
('2026-06', 2, 18100, 20000, 90.5)
[('Pathao', 5, 2.3), ('RedX', 4, 3.5), ('Steadfast', 3, 2.0)]

INSERT INTO orders (order_id, customer_id, order_date, status) VALUES (15, 8, '2026-06-20', 'pending');
INSERT INTO order_items (order_id, product_id, quantity, unit_price) VALUES (15, 1, 1, 1200), (15, 4, 2, 900);
UPDATE orders SET status = 'shipped' WHERE order_id = 13;
SELECT order_id, order_date, status FROM orders WHERE order_id >= 13 ORDER BY order_id;

+----------+------------+-----------+
| order_id | order_date | status    |
+----------+------------+-----------+
|       13 | 2026-06-02 | shipped   |
|       14 | 2026-06-09 | delivered |
|       15 | 2026-06-20 | pending   |
+----------+------------+-----------+

src = sqlite3.connect("shop.db")
newer_only = extract_order_lines(src, "2026-06-10")
print("without lookback:", sorted({r["order_id"] for r in newer_only}))
src.close()

run_pipeline(CourierAPI("courier_api.json"))

without lookback: [15]
INFO etl: incremental extract since 2026-05-26: 5 order lines, 12 deliveries, 6 targets
INFO etl: load: 2 inserted, 1 updated, 1 monthly rows changed; warehouse 22 lines, 106550 taka

# m  h  dom mon dow  command
15   2  *   *   *    cd /srv/shop-etl && .venv/bin/python run_etl.py >> logs/etl.log 2>&1

from datetime import datetime, timedelta
from airflow.sdk import DAG                       # Airflow 2-তে: from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator
# Airflow 2-তে: from airflow.operators.python import PythonOperator

with DAG(
    dag_id="shop_etl",
    schedule="15 2 * * *",                      # cron লাইনের মতোই
    start_date=datetime(2026, 6, 1),
    catchup=False,
    default_args={"retries": 3, "retry_delay": timedelta(minutes=5)},
) as dag:
    load_sales = PythonOperator(task_id="load_sales", python_callable=run_etl_main)
    build_features = PythonOperator(task_id="build_features", python_callable=build_features_main)
    load_sales >> build_features                # লোড সফল হলে তবেই features চলে

def health_check(wh_path, today, max_age_days=2):
    """সতর্কবার্তার লিস্ট ফেরত দেয়; খালি লিস্ট মানে সব ঠিক।"""
    wh = sqlite3.connect(wh_path)
    alerts = []
    last = wh.execute("SELECT status FROM etl_runs ORDER BY run_id DESC LIMIT 1").fetchone()
    if last is None or last[0] != "ok":
        alerts.append("last run did not succeed")
    newest = wh.execute("SELECT MAX(order_date) FROM fact_order_lines").fetchone()[0]
    age = (today - date.fromisoformat(newest)).days
    if age > max_age_days:
        alerts.append(f"newest order is {age} days old")
    wh.close()
    return alerts

print(sqlite3.connect("warehouse.db").execute(
    "SELECT run_id, mode, extracted, inserted, updated, status FROM etl_runs ORDER BY run_id").fetchall())
print(health_check("warehouse.db", today=date(2026, 6, 21)))
print(health_check("warehouse.db", today=date(2026, 6, 30)))

[(1, 'full', 20, 20, 0, 'ok'), (2, 'incremental', 3, 0, 0, 'ok'), (3, 'incremental', 5, 2, 1, 'ok')]
[]
['newest order is 10 days old']

import shutil

def test_transform_computes_delivery_days():
    lines = [{"order_id": 1, "product_id": 1, "order_date": "2026-01-12", "status": "delivered",
              "category": "Books", "quantity": 1, "revenue": 1200}]
    rows = transform(lines, [{"order_id": 1, "courier": "Pathao", "delivered_on": "2026-01-14"}])
    assert rows[0]["delivery_days"] == 2 and rows[0]["month"] == "2026-01"

def test_pipeline_is_idempotent():
    shutil.copy("shop.db", "test_shop.db")
    logging.disable(logging.CRITICAL)             # টেস্টের সময় লগ বন্ধ
    try:
        first = run_pipeline(CourierAPI("courier_api.json"), "test_shop.db", "test_wh.db", full=True)
        snapshot = fingerprint("test_wh.db")
        second = run_pipeline(CourierAPI("courier_api.json"), "test_shop.db", "test_wh.db", full=True)
    finally:
        logging.disable(logging.NOTSET)
    assert first[0] > 0 and second == (0, 0, 0)
    assert fingerprint("test_wh.db") == snapshot

test_transform_computes_delivery_days()
test_pipeline_is_idempotent()
print("pipeline tests passed")

pipeline tests passed

# ১. সহজ
wh = sqlite3.connect("warehouse.db")
print(wh.execute("""SELECT category, SUM(revenue) FROM fact_order_lines
                    WHERE status <> 'cancelled' GROUP BY category ORDER BY 2 DESC""").fetchall())
wh.close()

# ২. মাঝারি
def reconcile_by_status(src_path="shop.db", wh_path="warehouse.db"):
    src, wh = sqlite3.connect(src_path), sqlite3.connect(wh_path)
    source = dict(src.execute("""SELECT o.status, SUM(oi.quantity * oi.unit_price)
                                 FROM orders o JOIN order_items oi ON oi.order_id = o.order_id
                                 GROUP BY o.status""").fetchall())
    target = dict(wh.execute("SELECT status, SUM(revenue) FROM fact_order_lines GROUP BY status").fetchall())
    src.close()
    wh.close()
    if source != target:
        raise RuntimeError(f"status totals differ: {source} vs {target}")
    return target

print(reconcile_by_status())

# ৩. কঠিন
src = sqlite3.connect("shop.db")
with src:
    src.execute("UPDATE order_items SET quantity = 3 WHERE order_id = 15 AND product_id = 4")
src.close()
print(run_pipeline(CourierAPI("courier_api.json")))
print(sqlite3.connect("warehouse.db").execute(
    "SELECT * FROM monthly_sales WHERE month = '2026-06'").fetchone())
print(run_pipeline(CourierAPI("courier_api.json")))