Xây Dựng Data Pipeline Hiện Đại với Python và PostgreSQL

Python Stack: 3 Thư Viện Cốt Lõi

SQLAlchemy 2.0 với QueuePool và pool_pre_ping=True. Dùng COPY thay INSERT cho bulk loads — nhanh hơn 10-100x:

engine = create_engine(
    "postgresql+psycopg2://user:pass@localhost/db",
    pool_size=10, max_overflow=20,
    pool_pre_ping=True,
)

# FOR UPDATE SKIP LOCKED — bắt buộc khi chạy parallel workers
def fetch_unprocessed(limit=1000):
    with engine.connect() as conn:
        return conn.execute(text("""
            SELECT id, payload FROM raw_events
            WHERE processed = FALSE
            LIMIT :limit
            FOR UPDATE SKIP LOCKED
        """), {"limit": limit}).fetchall()

Pandas / Polars: pd.read_sql() đọc thẳng từ PostgreSQL. Polars + ConnectorX nhanh hơn Pandas 5-20x cho datasets lớn. Ghi về PostgreSQL dùng to_sql(..., method='multi', chunksize=5000).

Concurrency: 3 Pattern Phải Biết

Pattern 1 — ThreadPoolExecutor cho I/O-bound tasks (API calls, DB queries). max_workers=8 là sweet spot cho hầu hết pipeline.

Pattern 2 — asyncpg + asyncio cho high concurrency (500+ concurrent operations):

pool = await asyncpg.create_pool("postgresql://...", min_size=5, max_size=20)

async def fetch_events(batch_size=500):
    async with pool.acquire() as conn:
        async with conn.transaction():
            return await conn.fetch("""
                SELECT id, event_type, payload FROM raw_events
                WHERE processed = FALSE LIMIT $1
                FOR UPDATE SKIP LOCKED
            """, batch_size)

Pattern 3 — FOR UPDATE SKIP LOCKED: khi Worker 1 lock 100 rows, Worker 2 tự động SKIP qua — không deadlock, không duplicate processing. Đây là pattern quan trọng nhất khi chạy nhiều workers song song.

Orchestration: Prefect 3 vs Airflow

Prefect: @flow + @task decorators, retry tự động, UI built-in. Đủ dùng cho 95% startup pipeline.

Query Optimization: 5 Index Strategies

-- 1. Partial index — nhỏ, nhanh, chỉ index rows cần
CREATE INDEX idx_unprocessed ON raw_events (received_at)
  WHERE processed = FALSE;

-- 2. Composite index — thứ tự quan trọng (equality trước, range sau)
CREATE INDEX idx_user_time ON fact_user_events (user_id, event_at DESC);

-- 3. GIN index cho JSONB
CREATE INDEX idx_payload ON raw_events USING GIN (payload);

-- 4. Expression index cho computed values
CREATE INDEX idx_user_id_str ON raw_events ((payload->>'user_id'));

-- 5. BRIN cho time-series — rất nhỏ, hiệu quả cho append-only tables
CREATE INDEX idx_time_brin ON raw_events USING BRIN (received_at);

Dùng REFRESH MATERIALIZED VIEW CONCURRENTLY cho heavy aggregations — không lock đọc khi refresh.

Luôn debug với EXPLAIN (ANALYZE, BUFFERS). Chú ý: Seq Scan = thiếu index; Rows estimate sai xa = cần ANALYZE; cache hit ratio mục tiêu trên 95%.

Dead Letter Queue — Không Để Mất Records

Records lỗi không crash toàn batch — đẩy vào DLQ table với error_msg, attempt_count, last_failed_at. Alert khi DLQ tăng đột biến. Cho phép manual review và replay.

Khi Nào Dùng PostgreSQL (và Không Nên)

Production Checklist

• Partitioning theo thời gian cho bảng lớn
• Partial + GIN indexes phù hợp
• pool_pre_ping=True, PgBouncer khi nhiều connections
• FOR UPDATE SKIP LOCKED khi parallel workers
• Dead letter queue với alerting khi DLQ tăng đột biến
• Idempotent processing (chạy lại không tạo duplicate)
• pg_stat_statements enabled, cache hit ratio >95%

PostgreSQL + Python là stack thực dụng nhất cho startup đến mid-size. Khi data vượt 1TB hoặc cần Kafka-scale thì mới bổ sung thêm — đừng over-engineer từ đầu.

Muốn tích hợp AI vào data pipeline? Xem hướng dẫn xây dựng AI agents.