Python DE 1 — Python trong Data Engineering

Mở đầu: một ngày của data engineer trong ngân hàng

Sáng thứ Hai, bạn nhận được ba yêu cầu. Thứ nhất: kéo dữ liệu giao dịch thẻ từ một REST API của đối tác về mỗi giờ. Thứ hai: hợp nhất file CSV sao kê từ core banking với bảng khách hàng trong PostgreSQL, làm sạch, rồi nạp vào kho dữ liệu. Thứ ba: lên lịch để cả hai việc trên chạy tự động, có retry khi lỗi, có cảnh báo khi dữ liệu bất thường.

Ba yêu cầu này chạm vào ba trụ cột của data engineering — ingest, transform, orchestrate — và trong tuyệt đại đa số đội DE hiện nay, cả ba đều được viết bằng Python. Không phải vì Python là ngôn ngữ nhanh nhất (nó không nhanh), mà vì nó là chất keo (glue language) gắn kết mọi hệ thống lại với nhau: gọi API, đọc file, nói chuyện với database, biến đổi dữ liệu, rồi giao cho bộ điều phối chạy.

Bài này là bài mở đầu của series Python cho Data Engineering. Ta sẽ trả lời bốn câu hỏi lớn: (1) vì sao Python thống trị DE, (2) hệ sinh thái thư viện gồm những gì và mỗi thứ để làm gì, (3) quản lý môi trường và phụ thuộc sao cho tái lập được, (4) khi nào dùng Python và khi nào nên nhường chỗ cho SQL hoặc Spark. Cuối cùng là tư duy DE với Python và bản đồ các bài tiếp theo.

Series này giả định bạn đã biết Python cơ bản (nếu chưa, xem Python 7 — Pandas & NumPy cho phân tích dữ liệu). Ở đây ta nhìn Python dưới con mắt của người xây đường ống dữ liệu, không phải người phân tích một lần.

Vì sao Python thống trị data engineering

Có bốn lý do khiến Python trở thành ngôn ngữ mặc định của DE:

1. Glue language — chất keo vạn năng. Một pipeline điển hình phải nói chuyện với hàng chục hệ thống: HTTP API, S3/GCS, Kafka, Postgres, Snowflake, BigQuery, Redis... Python có thư viện client cho gần như mọi thứ. Bạn viết một script duy nhất kéo dữ liệu từ API, đẩy sang database, rồi kích hoạt một job Spark — mà không cần rời khỏi Python.

2. Hệ sinh thái phong phú. Cần đọc Parquet? pyarrow. Cần biến đổi dữ liệu bảng? pandas hoặc polars. Cần validate dữ liệu đầu vào? pydantic. Cần query SQL trong bộ nhớ? duckdb. Hiếm khi bạn phải tự viết lại thứ gì từ đầu.

3. Dễ đọc, dễ bảo trì. Code DE sống lâu và được nhiều người sửa. Cú pháp gần với tiếng Anh của Python giúp một người mới vào đội hiểu pipeline nhanh hơn nhiều so với, ví dụ, Scala hay Java.

4. Cộng đồng khổng lồ. Mọi lỗi bạn gặp gần như chắc chắn đã có người hỏi trên Stack Overflow. Mọi công cụ DE hiện đại (Airflow, dbt, Dagster, Prefect) đều có SDK Python hạng nhất.

Vai trò của Python trong pipeline

Hãy hình dung một pipeline batch chạy mỗi đêm. Python xuất hiện ở gần như mọi mắt xích:

• Ingest: lấy dữ liệu về. Nguồn có thể là API (httpx/requests), file trên object storage (s3fs, pyarrow), hay database khác (sqlalchemy). Xem chi tiết ở Kết nối DB.
• Transform: làm sạch, chuẩn hoá, join, tính toán. Đây là nơi pandas/polars/duckdb toả sáng.
• Load: ghi kết quả ra kho dữ liệu hoặc data lake (thường dạng Parquet).
• Orchestrate: gắn các bước lại thành lịch chạy có kiểm soát. Đây là địa hạt của Airflow — cũng viết bằng Python.

Hệ sinh thái thư viện: mỗi thứ để làm gì

Đây là "hộp đồ nghề" mà một DE Python dùng hằng ngày. Không cần thuộc lòng API — cần hiểu khi nào dùng cái nào.

Vài điểm cần nhấn mạnh:

pandas vs polars. pandas là tiêu chuẩn de facto suốt một thập kỷ, API rộng và tài liệu ngập tràn. Nhưng nó chạy đơn luồng và ngốn RAM. polars viết bằng Rust, chạy đa luồng, dùng định dạng cột Arrow, và quan trọng nhất là có lazy execution: nó nhìn toàn bộ pipeline trước khi chạy rồi tối ưu (đẩy filter và chọn cột xuống sát nguồn dữ liệu — predicate/projection pushdown). Trên dữ liệu thực tế, polars thường nhanh hơn pandas nhiều lần và dùng ít RAM hơn. Chi tiết ở Pandas deep-dive và Polars & Arrow.

pyarrow là nền móng thầm lặng. Arrow là định dạng bộ nhớ dạng cột mà polars, duckdb, Spark và pandas 2.x đều nói được. Vì cùng "ngôn ngữ", chúng có thể trao đổi dữ liệu không cần copy — đây là lý do Arrow trở thành lingua franca của thế giới dữ liệu.

pydantic — người gác cổng. DE thật sự bắt đầu từ giả định "dữ liệu đầu vào sẽ bẩn". pydantic giúp bạn khai báo schema mong đợi và tự động validate, thay vì để lỗi lan xuống tận cuối pipeline.

Quản lý môi trường & phụ thuộc

Đây là phần bị coi nhẹ nhất nhưng gây đau đầu nhất trong thực tế. Vấn đề cốt lõi: tái lập được (reproducibility). Pipeline chạy trên máy bạn hôm nay phải chạy y hệt trên server sản xuất tháng sau, với đúng phiên bản thư viện.

Vì sao không bao giờ cài global

Nếu pip install pandas thẳng vào Python hệ thống, mọi project chia sẻ chung một bộ thư viện. Project A cần pandas==1.5, project B cần pandas==2.2 — xung đột không thể giải quyết. Giải pháp là mỗi project một môi trường ảo (virtual environment) riêng biệt.

Các công cụ

• venv + pip: có sẵn trong Python, đơn giản, luôn tồn tại ở mọi nơi. Nhược điểm: pip chậm, và bạn phải tự quản lý lockfile.
• uv: công cụ thế hệ mới viết bằng Rust, nhanh hơn pip nhiều lần (nhiều benchmark báo 10–100x cho cài đặt, gấp khoảng 10 lần Poetry). Nó gộp vai trò của pip, virtualenv, và quản lý phiên bản Python vào một binary duy nhất, kèm lockfile universal. Đây là lựa chọn khuyến nghị cho project mới hiện nay.
• poetry: quản lý phụ thuộc khai báo qua pyproject.toml, có dependency groups và luồng publish lên PyPI mượt — phù hợp khi bạn phát hành thư viện.
• conda: mạnh khi cần thư viện có phần binary hệ thống (GDAL, CUDA...), phổ biến trong khoa học dữ liệu.

Lockfile & tái lập

pyproject.toml ghi ràng buộc mong muốn (ví dụ pandas >= 2.0). Lockfile (uv.lock, poetry.lock) ghi phiên bản chính xác đã được giải, kể cả phụ thuộc bắc cầu. Commit lockfile vào Git để mọi người và mọi server cài đúng một bộ.

# (minh hoạ) Khởi tạo project và thêm phụ thuộc bằng uv
uv init ncb-pipeline
cd ncb-pipeline
uv add polars pyarrow sqlalchemy pydantic httpx
uv run python pipeline.py        # chạy trong môi trường của project
uv sync                          # cài đúng theo uv.lock (trên server)

# (minh hoạ) Cách kinh điển với venv + pip
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install polars pyarrow sqlalchemy
pip freeze > requirements.txt    # "lockfile" thủ công

Trong ngân hàng, server thường nằm sau proxy CA nội bộ. Khi cài thư viện, hãy nhớ cấu hình chứng chỉ CA của công ty cho pip/uv nếu không sẽ lỗi SSL.

Khi nào dùng Python vs SQL vs Spark

Đây là quyết định kiến trúc quan trọng nhất. Chọn sai công cụ khiến pipeline vừa chậm vừa khó bảo trì. Nguyên tắc: để mỗi công cụ làm việc nó giỏi nhất.

Diễn giải:

• SQL trong warehouse: nếu dữ liệu đã nằm trong kho, đừng kéo hết về Python để xử lý. Warehouse được tối ưu để join và aggregate ngay tại chỗ. Đây là triết lý của dbt: biến đổi bằng SQL, ngay trong DW, có version control. Xem dbt.
• Python cho logic & glue: khi cần gọi API, xử lý điều kiện phức tạp, hay nối nhiều hệ thống — SQL bó tay, Python là nhà.
• Spark khi vượt một máy: khi dữ liệu lớn đến mức không một máy nào chứa nổi trong RAM, bạn cần xử lý phân tán. Đó là lúc chuyển sang Spark (nối tiếp series Spark).
• pandas/polars/duckdb cho tầm trung: phần lớn công việc DE hằng ngày rơi vào đây. Đừng vội dựng cluster Spark cho một file 2 GB — một máy với polars hoặc duckdb thừa sức.

Kinh nghiệm: 80% dữ liệu vừa với một máy tính hiện đại. Nhiều đội dựng Spark quá sớm rồi trả giá bằng chi phí vận hành. Bắt đầu bằng polars/duckdb, chỉ lên Spark khi thật sự cần.

Tư duy DE với Python

Viết script phân tích một lần rất khác viết pipeline chạy sản xuất mỗi đêm. Bốn tư duy cần khắc cốt:

1. Idempotency (chạy lại không sai)

Một pipeline tốt chạy hai lần phải cho cùng kết quả như chạy một lần. Nếu job đêm qua fail giữa chừng và bạn chạy lại, nó không được nhân đôi dữ liệu. Cách phổ biến: ghi theo partition (theo ngày) và ghi đè partition thay vì append mù quáng. Idempotency là điều kiện tiên quyết để retry an toàn — mà retry là chuyện thường ngày trong orchestration.

2. Batch vs stream

• Batch: xử lý theo lô, theo lịch (mỗi giờ, mỗi ngày). Đơn giản, dễ debug, phù hợp báo cáo. Phần lớn pipeline DE là batch.
• Stream: xử lý từng sự kiện gần thời gian thực (Kafka...). Phức tạp hơn nhiều, chỉ dùng khi độ trễ thật sự quan trọng (phát hiện gian lận tức thời chẳng hạn).

Đừng chọn stream chỉ vì nghe "hiện đại" — batch giải quyết hầu hết nhu cầu với chi phí thấp hơn nhiều.

3. Xử lý dữ liệu lớn hơn RAM

Khi file lớn hơn RAM, nạp hết một lúc sẽ sập. Hai kỹ thuật:

• Chunking: đọc theo từng khối (pandas chunksize), xử lý rồi ghi, giải phóng bộ nhớ.
• Lazy execution: với polars, xây pipeline dạng "kế hoạch" rồi để engine đọc và xử lý theo luồng (streaming), chỉ giữ phần cần thiết trong RAM.

# (minh hoạ) Đọc file lớn theo khối với pandas
import pandas as pd

tong_theo_chi_nhanh = {}
for chunk in pd.read_csv("giao_dich.csv", chunksize=500_000):
    g = chunk.groupby("chi_nhanh")["so_tien"].sum()
    for cn, tien in g.items():
        tong_theo_chi_nhanh[cn] = tong_theo_chi_nhanh.get(cn, 0) + tien

# (minh hoạ) Lazy scan với polars — không nạp hết vào RAM
import polars as pl

ket_qua = (
    pl.scan_csv("giao_dich.csv")            # lazy: chưa đọc gì
      .filter(pl.col("so_tien") > 500_000_000)
      .group_by("chi_nhanh")
      .agg(pl.col("so_tien").sum().alias("tong"))
      .collect()                            # đến đây engine mới chạy & tối ưu
)

4. Cấu trúc code

Script một-file dài 500 dòng là bom hẹn giờ. Pipeline sản xuất cần: tách hàm nhỏ (extract / transform / load riêng), cấu hình tách khỏi logic, logging tử tế, xử lý lỗi rõ ràng, và test. Đây là chủ đề của Production quality.

Use case thực tế: pipeline sao kê hằng ngày

Đội DE ngân hàng cần một pipeline chạy 2h sáng mỗi ngày:

1. Ingest: httpx kéo tỷ giá từ API đối tác; pyarrow/s3fs đọc file sao kê Parquet mà core banking đẩy lên object storage; sqlalchemy lấy bảng khách hàng từ PostgreSQL.
2. Transform: pydantic validate schema file sao kê (chặn dòng thiếu mã KH); polars (lazy) join sao kê với khách hàng, tính tổng biến động theo chi nhánh, gắn cờ giao dịch trên 500 triệu.
3. Load: ghi kết quả ra Parquet trên data lake theo partition ngày (idempotent), đồng thời nạp bảng tổng hợp vào warehouse cho BI.
4. Orchestrate: một DAG Airflow lên lịch, retry 3 lần nếu API chập chờn, gửi alert Slack nếu số dòng lệch bất thường.

Toàn bộ chạy trên một máy vừa phải — không cần Spark, vì dữ liệu một ngày chỉ vài GB. Môi trường được đóng gói bằng uv với uv.lock commit trong Git, đảm bảo server sản xuất cài đúng phiên bản như máy dev.

Bản đồ series

Các bài tiếp theo sẽ đào sâu từng mảnh:

• Pandas deep-dive — làm chủ pandas cho DE: chỉ số, groupby nâng cao, hiệu năng.
• Polars & Arrow — DataFrame thế hệ mới, lazy execution, Parquet.
• Kết nối DB — sqlalchemy, connection pool, đọc/ghi database hiệu quả.
• Production quality — cấu trúc code, logging, test, đóng gói pipeline.

Và các series liên quan: Airflow cho orchestration, dbt cho transform bằng SQL trong warehouse.

Ghi nhớ

• Python là chất keo của DE: nó không nhanh nhất nhưng nối được mọi hệ thống, có hệ sinh thái phong phú và cộng đồng khổng lồ.
• Ba trụ cột: ingest → transform → load, tất cả được điều phối (orchestrate) bởi công cụ như Airflow — cũng viết bằng Python.
• Hộp đồ nghề: pandas/polars (transform), pyarrow (Arrow/Parquet, ngôn ngữ chung), sqlalchemy (DB), httpx (API), pydantic (validate), duckdb (SQL trong tiến trình).
• Luôn dùng môi trường ảo, commit lockfile. uv là lựa chọn hiện đại và nhanh cho project mới; venv+pip luôn có sẵn; poetry mạnh khi phát hành thư viện.
• Chọn đúng công cụ: SQL cho transform trong warehouse, Python cho logic/glue, Spark khi vượt một máy, polars/duckdb cho phần lớn công việc tầm trung. Đừng dựng Spark quá sớm.
• Tư duy sản xuất: idempotency để retry an toàn, ưu tiên batch, xử lý dữ liệu lớn hơn RAM bằng chunk/lazy, và cấu trúc code sạch để pipeline sống lâu.