Python DE 5 — Kết nối & nạp dữ liệu vào cơ sở dữ liệu

Với một data engineer, database vừa là nguồn (extract từ hệ thống nghiệp vụ) vừa là đích (load vào warehouse). Viết được câu SELECT thì ai cũng làm được; nhưng nạp 10 triệu dòng mà pipeline không sập, không nuốt hết RAM, chạy lại được mà không nhân đôi dữ liệu — đó mới là chuyện của production. Bài này đi từ driver nền tảng lên SQLAlchemy, rồi tập trung vào phần quan trọng nhất: ghi dữ liệu lớn thật nhanh và thật an toàn.

Bài này nối tiếp pandas (dữ liệu sau khi transform nằm trong DataFrame) và File & I/O (đọc file trước khi nạp vào DB).

1. Driver & thư viện: các tầng trừu tượng

Python nói chuyện với database qua nhiều tầng. Hiểu tầng nào làm gì giúp bạn chọn công cụ đúng.

Tầng 1 — DB-API 2.0 (PEP 249). Đây là chuẩn giao diện mà mọi driver Python tuân theo. Bạn luôn gặp cùng bộ khái niệm: connection, cursor, execute(), fetchone/fetchmany/fetchall, commit(), rollback(). Học một driver là gần như biết dùng tất cả.

Tầng 2 — driver theo từng loại DB. Mỗi database có driver riêng, viết bằng C hoặc Python:

Tầng 3 — SQLAlchemy: lớp trừu tượng. SQLAlchemy ngồi trên driver và cho bạn một API thống nhất, độc lập với DB cụ thể. Nó có hai phong cách:

• Core: làm việc ở mức "SQL có kiểu" — bạn vẫn nghĩ theo bảng, cột, câu lệnh. Đây là lựa chọn phù hợp nhất cho data engineering (ETL/ELT, bulk load).
• ORM: ánh xạ bảng thành class Python. Tiện cho ứng dụng web CRUD, nhưng thường thừa và chậm cho pipeline nạp hàng triệu dòng.

Với DE, hãy mặc định dùng Core. SQLAlchemy quản lý hai đối tượng chính:

• Engine: điểm khởi đầu, đại diện cho một pool kết nối tới một DB. Tạo một lần cho cả ứng dụng.
• Connection: một kết nối lấy ra từ pool để chạy lệnh, thường dùng qua context manager.

2. Kết nối an toàn

2.1. Connection string & KHÔNG hardcode credential

SQLAlchemy dùng URL dạng dialect+driver://user:password@host:port/dbname:

# minh hoạ — postgresql + psycopg v3
from sqlalchemy import create_engine
import os

url = (
    f"postgresql+psycopg://{os.environ['DB_USER']}:{os.environ['DB_PASS']}"
    f"@{os.environ['DB_HOST']}:5432/{os.environ['DB_NAME']}"
)
engine = create_engine(url)

Quy tắc số một: không bao giờ hardcode username/password vào code. Credential đi vào code sẽ theo Git lên repo, lộ ra ngay. Lấy chúng từ:

• biến môi trường (os.environ) cho môi trường container/CI;
• secret manager (Vault, AWS Secrets Manager, GCP Secret Manager) cho production;
• file .env chỉ dùng khi dev, và luôn nằm trong .gitignore.

Với dữ liệu đi qua mạng công cộng, hãy bật SSL/TLS. Với psycopg bạn truyền qua connect_args:

# minh hoạ
engine = create_engine(url, connect_args={"sslmode": "require"})

2.2. Connection pooling: vì sao và cấu hình ra sao

Mở một kết nối TCP tới database tốn kém: bắt tay TCP, xác thực, TLS handshake, cấp phát tài nguyên phía server. Nếu mỗi query lại mở rồi đóng một kết nối, phần lớn thời gian trôi vào việc thiết lập chứ không phải chạy SQL.

Connection pool giải quyết điều này: giữ sẵn một số kết nối mở, cho code "mượn" (checkout) rồi "trả lại" (checkin) thay vì đóng thật. SQLAlchemy mặc định dùng QueuePool. Các tham số truyền thẳng vào create_engine:

# minh hoạ
engine = create_engine(
    url,
    pool_size=5,        # số kết nối giữ thường trực trong pool
    max_overflow=10,    # số kết nối "tràn" thêm khi cao điểm
    pool_timeout=30,    # giây chờ tối đa để lấy một kết nối
    pool_recycle=1800,  # tái tạo kết nối sau 30 phút (tránh bị server đóng)
    pool_pre_ping=True, # ping trước khi dùng, thay kết nối đã chết
)

Vài điểm theo tài liệu chính thức:

• pool_size mặc định 5. Pool không mở sẵn kết nối — nó chỉ tạo khi có yêu cầu đầu tiên. Tổng số kết nối tối đa ra DB là pool_size + max_overflow.
• pool_pre_ping=True giúp xử lý êm các kết nối đã bị server đóng (rất hữu ích khi DB restart hoặc idle lâu).
• Không chia sẻ pool qua tiến trình fork. Kết nối TCP là file descriptor; hai process cùng dùng một socket sẽ hỏng dữ liệu. Nếu dùng multiprocessing, tạo engine sau khi fork trong từng process con.
• QueuePool không dùng với asyncio; create_async_engine() tự chọn AsyncAdaptedQueuePool.

Cỡ pool nên bắt đầu nhỏ (5 + 10) rồi tăng dần theo quan sát. Đặt pool quá lớn có thể làm cạn số kết nối tối đa của Postgres (max_connections).

2.3. Đóng kết nối đúng cách

Luôn dùng context manager để kết nối được trả về pool ngay cả khi có lỗi:

# minh hoạ
from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("SELECT count(*) FROM orders"))
    print(result.scalar())
# ra khỏi block: kết nối tự động trả về pool

Quên đóng kết nối là nguyên nhân số một của lỗi kinh điển QueuePool limit ... connection timed out: pool cạn vì kết nối bị "mượn" mà không bao giờ trả.

3. Đọc dữ liệu

3.1. Chỉ lấy đúng thứ cần

Trước mọi tối ưu, hãy nhớ nguyên tắc rẻ nhất: chỉ SELECT cột cần, chỉ lấy dòng cần. Đừng SELECT * rồi lọc trong pandas — hãy đẩy WHERE, LIMIT, cả phép tổng hợp xuống DB. Database được sinh ra để làm việc đó.

3.2. pd.read_sql và đọc theo chunk

Nạp cả bảng lớn vào một DataFrame là cách nhanh nhất để hết RAM. Dùng chunksize để nhận về một iterator các DataFrame nhỏ:

# minh hoạ
import pandas as pd

for chunk in pd.read_sql("SELECT id, amount FROM orders", engine, chunksize=50_000):
    process(chunk)   # xử lý từng 50k dòng

3.3. Cạm bẫy: chunksize KHÔNG đủ — cần server-side cursor

Đây là chi tiết nhiều người bỏ sót. Với mặc định, driver kéo toàn bộ kết quả về client rồi mới cắt thành chunk — RAM vẫn nổ. Muốn thật sự streaming, phải bật server-side cursor để DB chỉ gửi từng lô dòng khi được yêu cầu:

# minh hoạ
with engine.connect().execution_options(stream_results=True) as conn:
    for chunk in pd.read_sql("SELECT * FROM big_events", conn, chunksize=50_000):
        process(chunk)

stream_results=True yêu cầu SQLAlchemy dùng server-side cursor; kết hợp với chunksize, bộ nhớ client giữ ổn định dù bảng có hàng trăm triệu dòng. (Polars cũng đọc được qua pl.read_database, nhưng nguyên tắc streaming là như nhau.)

4. Ghi dữ liệu hiệu quả — phần quan trọng nhất

4.1. Vì sao to_sql chậm

DataFrame.to_sql() tiện nhưng mặc định sinh ra một câu INSERT cho mỗi dòng (hoặc mỗi nhóm nhỏ), mỗi câu là một vòng round-trip client↔server. Với 10 triệu dòng, đó là hàng triệu round-trip — có thể mất hàng chục phút tới hàng giờ. to_sql(..., method='multi') gộp nhiều dòng vào một INSERT nên nhanh hơn đáng kể, nhưng vẫn không đọ được với COPY.

Thứ tự tốc độ (từ chậm tới nhanh) trên Postgres, thực nghiệm phổ biến:

1. to_sql mặc định (INSERT từng dòng) — chậm nhất.
2. executemany — nhanh hơn chút.
3. to_sql(method='multi') / execute_values — gộp nhiều dòng mỗi câu, nhanh đáng kể.
4. COPY FROM STDIN — nhanh nhất, thường nhanh hơn INSERT hàng chục lần.

4.2. COPY: đường nhanh của PostgreSQL

COPY là lệnh bulk load gốc của Postgres, bỏ qua chi phí phân tích từng câu INSERT. Với psycopg v3, dùng context manager cursor.copy():

# minh hoạ — psycopg v3, nạp DataFrame bằng COPY
import io
import csv
import psycopg

def copy_dataframe(df, table, cols, dsn):
    buf = io.StringIO()
    df.to_csv(buf, index=False, header=False, quoting=csv.QUOTE_MINIMAL)
    buf.seek(0)
    collist = ", ".join(cols)
    with psycopg.connect(dsn) as conn:          # transaction bao trọn
        with conn.cursor() as cur:
            with cur.copy(
                f"COPY {table} ({collist}) FROM STDIN WITH (FORMAT csv)"
            ) as copy:
                copy.write(buf.read())
        conn.commit()

Trong psycopg v3, nếu có sẵn iterable các tuple, bạn có thể copy.write_row(row) từng dòng mà không cần dựng CSV. Nếu source và target có schema y hệt, FORMAT BINARY còn nhanh hơn nữa (nhưng khắt khe về kiểu — server không tự ép int4 → int8).

Nếu vẫn muốn dùng to_sql cho tiện nhưng nhanh, có thể gắn một callable method= gọi COPY phía dưới. Nhưng với dữ liệu thật lớn, gọi COPY trực tiếp là rõ ràng và dễ kiểm soát nhất.

4.3. Staging table + INSERT ... SELECT

Một mẫu kinh điển trong warehouse: đừng COPY thẳng vào bảng đích (có index, ràng buộc, trigger — mọi thứ làm chậm insert). Thay vào đó:

1. COPY dữ liệu thô vào một staging table trống, không index.
2. Làm sạch / khử trùng / join ngay trong SQL.
3. INSERT INTO target SELECT ... FROM staging (hoặc upsert) trong một transaction.

Cách này tách "nạp nhanh" khỏi "biến đổi", và cho bạn điểm dừng an toàn để kiểm tra trước khi chạm vào bảng đích.

4.4. Upsert theo batch (INSERT ... ON CONFLICT)

Khi cần "có thì update, chưa có thì insert", Postgres dùng INSERT ... ON CONFLICT (từ 9.5). Đây cũng là chìa khoá cho incremental load idempotent:

# minh hoạ — upsert theo batch với psycopg v3
import psycopg
from psycopg.rows import tuple_row

UPSERT = """
    INSERT INTO dim_customer (customer_id, name, updated_at)
    VALUES (%s, %s, %s)
    ON CONFLICT (customer_id)
    DO UPDATE SET
        name       = EXCLUDED.name,
        updated_at = EXCLUDED.updated_at
"""

def upsert_batch(rows, dsn):
    with psycopg.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.executemany(UPSERT, rows)   # rows: list các tuple
        conn.commit()

Vài lưu ý theo tài liệu Postgres:

• Bảng ảo EXCLUDED chứa các giá trị lẽ ra được insert nhưng bị đụng độ — dùng nó để cập nhật.
• DO NOTHING rẻ hơn DO UPDATE khi bạn chỉ cần "đảm bảo dòng tồn tại" (kiểu idempotency key).
• Cần một unique constraint / primary key trên cột conflict thì ON CONFLICT mới hoạt động.
• Với tốc độ ghi rất cao lên cùng một key, nên gộp/debounce trước ở phía Python thay vì bắn stream upsert đơn dòng (chúng serialize trên cùng một lock).

Chi tiết cú pháp UPSERT xem SQL nâng cao trên Postgres.

5. Transaction & an toàn dữ liệu

Atomicity là hợp đồng của transaction: hoặc tất cả thay đổi được ghi, hoặc không gì cả. Nếu bước 3 của việc nạp thất bại, các bước 1–2 phải được rollback để không để lại dữ liệu nửa vời.

# minh hoạ — SQLAlchemy: begin() tự commit khi thành công, rollback khi lỗi
with engine.begin() as conn:
    conn.execute(text("TRUNCATE staging"))
    conn.execute(text("INSERT INTO staging SELECT ... "))
    conn.execute(text("INSERT INTO target SELECT ... FROM staging"))
# thoát block không lỗi -> commit; có exception -> rollback toàn bộ

Lưu ý phân biệt: engine.connect() cho bạn kết nối và bạn tự commit/rollback; còn engine.begin() mở sẵn một transaction và tự commit/rollback theo kết quả block. Với ETL, begin() thường an toàn hơn.

Idempotent load — nạp lại không nhân đôi dữ liệu. Pipeline sẽ chạy lại (retry, backfill), nên mỗi lần chạy phải cho cùng một kết quả. Hai chiến lược chính:

• Truncate–load: xoá sạch phân vùng/bảng rồi nạp lại toàn bộ. Đơn giản, hợp với bảng nhỏ/trung bình hoặc snapshot theo ngày.
• Incremental + upsert: chỉ nạp phần thay đổi, dùng ON CONFLICT để hợp nhất theo key. Hiệu quả với bảng lớn, nhưng đòi hỏi key ổn định.

Tránh INSERT thuần cho incremental — chạy lại là nhân đôi ngay.

6. Parameterized query & chống SQL injection

Quy tắc không bao giờ phá vỡ: đừng bao giờ nối chuỗi để dựng SQL từ dữ liệu bên ngoài. Luôn dùng bind parameter để driver tự escape:

# minh hoạ
# SAI — lỗ hổng SQL injection
conn.execute(text(f"SELECT * FROM users WHERE name = '{user_input}'"))

# ĐÚNG — bind parameter (SQLAlchemy dùng :name)
conn.execute(text("SELECT * FROM users WHERE name = :n"), {"n": user_input})

# ĐÚNG — DB-API dùng placeholder của driver (psycopg: %s)
cur.execute("SELECT * FROM users WHERE name = %s", (user_input,))

Bind param còn cho phép DB cache execution plan cho câu lệnh lặp lại — vừa an toàn vừa nhanh hơn.

7. Use case thực tế: nạp 10 triệu dòng bằng COPY

Một job hàng đêm phải nạp ~10 triệu dòng giao dịch từ file CSV vào bảng fact_transactions trên Postgres.

Cách sai (đã gặp): df.to_sql("fact_transactions", engine, if_exists="append"). Chạy hơn một giờ, chiếm cả pool kết nối, thỉnh thoảng timeout.

Cách đúng:

1. Đọc CSV theo chunk (không nạp cả file vào RAM — xem File & I/O).
2. COPY từng chunk vào staging table trống, không index.
3. Chạy kiểm tra chất lượng ngay trong SQL (đếm dòng, null key).
4. INSERT INTO fact_transactions SELECT ... FROM staging trong một transaction; hoặc upsert nếu là incremental.
5. TRUNCATE staging để lần sau sạch.

Kết quả điển hình: từ hơn một giờ xuống còn vài phút, RAM phẳng, và job có thể chạy lại an toàn. Sau khi nạp khối lớn, chạy ANALYZE để cập nhật thống kê cho planner — chi tiết bảo trì và COPY xem Bảo trì DBA trên Postgres.

8. Khi nào dùng Python, khi nào dùng công cụ chuyên

Python + SQLAlchemy là "dao Thụy Sĩ" tuyệt vời cho extract → transform nhẹ → load, đặc biệt khi logic phức tạp hoặc cần gọi API xen kẽ. Nhưng nó không phải câu trả lời cho mọi thứ:

• Transform trong warehouse (SQL thuần, versioned, test được): dùng dbt.
• Dữ liệu quá lớn cho một máy (hàng TB, cần phân tán): dùng Spark.
• Bắt thay đổi realtime từ DB nguồn: dùng CDC (Debezium…) thay vì poll bằng Python.

Xem thêm cách đóng gói job Python chuẩn production ở Chất lượng & Production.

Ghi nhớ

• SQLAlchemy Core + psycopg là combo mặc định cho DE trên Postgres; tạo Engine một lần, dùng qua context manager.
• Credential lấy từ env/secret, không hardcode; bật SSL trên mạng công cộng.
• Connection pool vì mở kết nối rất tốn; cấu hình pool_size, max_overflow, pool_pre_ping; luôn trả kết nối về pool.
• Đọc lớn: chunksize + stream_results=True (server-side cursor), và chỉ SELECT thứ cần.
• Ghi lớn: to_sql mặc định là chậm nhất → dùng COPY (nhanh nhất), staging table, method='multi'/execute_values khi không có COPY.
• Upsert bằng INSERT ... ON CONFLICT để load incremental idempotent.
• Transaction bao trọn để atomic; thiết kế load chạy lại không nhân đôi (truncate-load hoặc upsert).
• Bind parameter luôn luôn — không nối chuỗi SQL.

Sources:

• SQLAlchemy Connection Pooling
• psycopg 3 — Using COPY
• PostgreSQL — INSERT ... ON CONFLICT
• Loading SQL data into Pandas without running out of memory
• Fastest Way to Load Data Into PostgreSQL Using Python — Haki Benita