Một pipeline dữ liệu tốt không phải là một script dài chạy được một lần, mà là chuỗi bước nhỏ, mỗi bước có đầu vào, đầu ra và cách kiểm tra riêng, chạy lại bao nhiêu lần cũng cho cùng kết quả. Bài này dựng trọn một pipeline dữ liệu giá sản phẩm: thu thập HTML bằng httpx, làm sạch bằng BeautifulSoup và pandas, nạp vào SQLite theo kiểu idempotent, phân tích bằng Polars rồi hiển thị trên dashboard Streamlit. Cuối bài có kiểm thử tự động cho cả dashboard và danh sách lỗi hay gặp khi đưa pipeline dữ liệu vào chạy định kỳ.
Bài ghép lại những mảnh đã viết riêng trong series: crawl bằng requests và BeautifulSoup, crawler lịch sự và hợp lệ, làm sạch dữ liệu bằng pandas và Polars cho dữ liệu lớn. Mọi số liệu là output thật trên Apple M4, 16 GB, Python 3.12 (phiên bản gói ghi ở requirements.txt), mã nguồn ở code/s7-54. Dữ liệu thật chỉ là 100 sách của trang thử nghiệm; phần lịch sử giá được mô phỏng và được đánh dấu rõ trong cơ sở dữ liệu.
Bài toán của pipeline dữ liệu giá sản phẩm
Một cửa hàng muốn theo dõi giá 100 sản phẩm theo ngày để trả lời ba câu hỏi: mặt bằng giá đang tăng hay giảm, sản phẩm nào đang rẻ bất thường so với lịch sử của chính nó, và còn hàng hay không. Với doanh nghiệp bán lẻ, đó là bài toán giám sát giá đối thủ hoặc giá nhà cung cấp; ở đây ta dùng books.toscrape.com, trang do Zyte dựng riêng cho việc học crawl, để ví dụ chạy lại được và không động tới trang của ai.
Điểm quan trọng nhất của pipeline dữ liệu là chia bài toán thành các tầng có ranh giới rõ, vì chính ranh giới đó giúp sửa lỗi nhanh:
| Tầng | Việc làm | Lưu ở đâu | Vì sao tách riêng |
|---|---|---|---|
| 1. Thu thập | Tải HTML, không diễn giải | data/raw/<ngày>/page-N.html | Giữ bản gốc; parser sai thì chạy lại từ file, không tải lại |
| 2. Biến đổi | Parse, làm sạch, kiểm tra chất lượng | DataFrame trong bộ nhớ | Dữ liệu xấu bị chặn trước khi vào DB |
| 3. Nạp | Ghi vào SQLite bằng upsert | data/prices.db | Chạy lại không nhân đôi dòng |
| 4. Phân tích | SQL + Polars tính chỉ số | DataFrame Polars | Logic tính toán tách khỏi giao diện |
| 5. Hiển thị | Streamlit đọc DB, vẽ biểu đồ | Trình duyệt | Giao diện chỉ đọc, không ghi DB |
Mỗi file của pipeline dữ liệu làm một việc: config.py, crawl.py, transform.py, load.py, analyze.py, app.py, và run_pipeline.py để nối các bước.
Tầng 1: thu thập lịch sự và lưu bản gốc
Nguyên tắc đầu tiên của tầng thu thập trong pipeline dữ liệu là không diễn giải gì cả. Ta tải HTML và ghi nguyên văn ra đĩa theo thư mục ngày. Nếu parser đọc sai rating, ta sửa parser rồi xử lý lại các trang đã lưu, không cần gọi lại máy chủ. Tầng này bất biến: chỉ thêm, không sửa.
def fetch(client: httpx.Client, url: str, retries: int = 3) -> str:
for attempt in range(1, retries + 1):
try:
r = client.get(url)
r.raise_for_status()
return r.text
except httpx.HTTPError as exc:
wait = 2 ** attempt
print(f" lỗi {exc!r} (lần {attempt}/{retries}), chờ {wait}s", file=sys.stderr)
time.sleep(wait)
raise RuntimeError(f"không tải được {url}")Ba chi tiết đáng chú ý. Thứ nhất, timeout=15 được đặt tường minh: theo tài liệu HTTPX, mặc định là 5 giây không có dữ liệu, và có bốn loại timeout khác nhau (connect, read, write, pool), nên một crawler chạy lâu nên chọn giá trị có chủ đích thay vì để mặc định. Thứ hai, thử lại với thời gian chờ tăng gấp đôi (2, 4, 8 giây) để không dồn dập vào máy chủ. Thứ ba, raise_for_status() biến mã 404 hoặc 500 thành ngoại lệ; thiếu dòng này, trang lỗi sẽ được lưu như thể là dữ liệu hợp lệ và hỏng ngầm ở tầng sau.
Về tính hợp lệ: ta đặt User-Agent nêu rõ tên và mục đích, nghỉ 1 giây giữa hai request (DELAY_S = 1.0) và chỉ tải 5 trang. Trang books.toscrape.com không có robots.txt (máy chủ trả 404 khi kiểm tra) và được dựng để thực hành crawl, nên đây là ví dụ an toàn. Với trang thật, hãy đọc robots.txt, điều khoản sử dụng, giới hạn tốc độ và tránh thu thập dữ liệu cá nhân; chi tiết đã nói ở bài crawler lịch sự và hợp lệ. Nếu cần crawl hàng nghìn trang có hàng đợi và retry, hãy chuyển sang Scrapy; vòng lặp trên chỉ hợp cho vài chục trang.
Tầng 2: biến đổi và kiểm tra chất lượng
Hàm parse_page đọc từng thẻ article.product_pod, ta lấy khoá, tên, chuỗi giá, rating và tình trạng kho. Khoá tự nhiên là slug lấy từ đường dẫn (ví dụ a-light-in-the-attic_1000), vì tên sách có thể đổi còn đường dẫn thì ổn định hơn nhiều. Khoá đúng là nền tảng của upsert ở tầng sau.
Giá trên trang là chuỗi £51.77. Nếu trang được giải mã sai bảng mã, ký tự đó thành £, một lỗi kinh điển khi crawl. Thay vì xử lý riêng từng trường hợp, ta bỏ mọi ký tự không phải số hoặc dấu chấm bằng một biểu thức chính quy rồi ép kiểu float; cách này chịu được cả hai dạng. Việc làm sạch chi tiết hơn (khử trùng lặp, kiểu dữ liệu, giá trị thiếu) nằm ở bài làm sạch dữ liệu crawl bằng pandas.
df["price"] = df["price_text"].str.replace(r"[^0-9.]", "", regex=True).astype(float)
df = df.drop(columns="price_text").drop_duplicates("slug", keep="last")Phần quan trọng nhất của tầng biến đổi trong pipeline dữ liệu là hàm validate: nó trả về danh sách lỗi, rỗng nghĩa là đạt. Ta kiểm tra bảng không rỗng, giá không thiếu và dương, rating nằm trong 1 đến 5, không trùng khoá. Nếu có lỗi, pipeline dữ liệu dừng bằng ngoại lệ trước khi chạm vào cơ sở dữ liệu. Nguyên tắc là "hỏng thì dừng to": dữ liệu sai âm thầm trong DB tệ hơn nhiều so với một lần chạy thất bại có thông báo.
Ví dụ: khi trang đổi giao diện, bộ chọn CSS không khớp và parse_page trả về 0 dòng; thiếu validate, ta ghi một ngày dữ liệu rỗng mà không ai biết. Vì thế clean luôn trả bảng có đủ cột, kể cả khi không có dòng nào, để validate báo lỗi "không có dòng nào" thay vì một KeyError khó hiểu.
Tầng 3: nạp vào SQLite theo kiểu idempotent
Idempotent nghĩa là chạy hai lần cho cùng kết quả như chạy một lần. Đây là đặc tính bắt buộc của mọi pipeline dữ liệu chạy định kỳ, vì job nào rồi cũng có lúc bị chạy lại do mất mạng, bấm nhầm hoặc chạy bù. Lược đồ dùng khoá chính ghép (slug, day), nên mỗi sản phẩm chỉ có đúng một giá trong một ngày:
CREATE TABLE IF NOT EXISTS price_history (
slug TEXT NOT NULL REFERENCES products(slug),
day TEXT NOT NULL,
price REAL NOT NULL CHECK (price > 0),
in_stock INTEGER NOT NULL,
source TEXT NOT NULL, -- 'crawl' hoặc 'simulated'
PRIMARY KEY (slug, day)
);Bước ghi của pipeline dữ liệu dùng INSERT ... ON CONFLICT DO UPDATE. Theo tài liệu SQLite, cú pháp upsert xuất hiện từ phiên bản 3.24.0 (2018), theo mẫu của PostgreSQL; tiền tố excluded. trỏ tới giá trị mà lệnh INSERT định ghi. Một điểm cần nhớ: chỉ ràng buộc duy nhất (khoá chính, UNIQUE) mới kích hoạt upsert, còn CHECK hay NOT NULL bị vi phạm thì vẫn báo lỗi và lệnh bị huỷ.
with con: # một transaction: hoặc ghi hết, hoặc không ghi gì
con.executemany(
"INSERT INTO price_history(slug, day, price, in_stock, source) VALUES (?,?,?,?,?) "
"ON CONFLICT(slug, day) DO UPDATE SET price=excluded.price, "
"in_stock=excluded.in_stock, source=excluded.source",
[(r.slug, day, float(r.price), int(r.in_stock), source) for r in df.itertuples()])Trong pipeline dữ liệu, khối with con: bọc cả đợt ghi trong một transaction, nên nếu dòng thứ 80 vi phạm CHECK (price > 0), 79 dòng trước đó cũng được hoàn tác. SQLite phù hợp khi pipeline dữ liệu chạy trên một máy và dashboard chỉ đọc; khi nhiều tiến trình ghi đồng thời hoặc dữ liệu lên hàng chục GB, hãy chuyển sang PostgreSQL với cùng cú pháp upsert, phần còn lại của mã gần như giữ nguyên.
Chạy cả pipeline dữ liệu và đọc output thật
Script run_pipeline.py của pipeline dữ liệu nối bốn bước đầu. Vì trang thử nghiệm không đổi giá theo ngày, ta lấy giá thật của lần crawl làm mốc (nhãn ngày đặt tay là 2026-07-01 để kết quả tái lập), rồi sinh 29 ngày lịch sử giá giả lập với seed cố định: giá dao động quanh mốc khoảng 2% và 15% sản phẩm có một đợt giảm giá 10 đến 30% kéo dài 3 đến 6 ngày. Mọi dòng giả lập mang source = 'simulated' để không ai nhầm với dữ liệu thật.
$ python run_pipeline.py --fresh
[1/4] crawl
đã lưu raw/2026-07-01/page-1.html (50,449 byte)
đã lưu raw/2026-07-01/page-2.html (50,853 byte)
đã lưu raw/2026-07-01/page-3.html (51,354 byte)
đã lưu raw/2026-07-01/page-4.html (52,504 byte)
đã lưu raw/2026-07-01/page-5.html (51,809 byte)
[2/4] transform
100 sản phẩm, giá 10.16-58.11 GBP, rating TB 2.93
[3/4] load
price_history: 3000 dòng, 100 sản phẩm, 30 ngày
[4/4] analyze
shape: (2, 4)
┌─────────────────────────────────┬────────┬────────┬───────────────┐
│ title ┆ median ┆ latest ┆ vs_median_pct │
│ --- ┆ --- ┆ --- ┆ --- │
│ str ┆ f64 ┆ f64 ┆ f64 │
╞═════════════════════════════════╪════════╪════════╪═══════════════╡
│ Throwing Rocks at the Google B… ┆ 31.225 ┆ 22.2 ┆ -28.9 │
│ My Paris Kitchen: Recipes and … ┆ 33.35 ┆ 25.84 ┆ -22.5 │
└─────────────────────────────────┴────────┴────────┴───────────────┘
tổng thời gian: 9.5sĐọc kết quả của pipeline dữ liệu: 5 trang cho 100 sản phẩm, đủ 3.000 dòng (100 sản phẩm nhân 30 ngày), trong đó 100 dòng có nguồn crawl và 2.900 dòng simulated. Phần lớn 9,5 giây là do ta chủ động nghỉ 1 giây giữa các trang. Chỉ có 2 sản phẩm đang rẻ hơn trung vị lịch sử của chính chúng từ 10% trở lên: đó là hai sản phẩm có đợt giảm giá giả lập vẫn đang chạy ở ngày cuối cùng.
Tầng 4: phân tích bằng SQL và Polars
Tầng phân tích của pipeline dữ liệu đọc từ DB bằng một câu SQL JOIN, rồi tính chỉ số bằng Polars. Với 3.000 dòng, pandas cũng đủ nhanh; ta chọn Polars vì cú pháp biểu thức dễ đọc khi gộp nhóm và dữ liệu sẽ lớn dần (so sánh hai thư viện ở bài Polars cho dữ liệu lớn). Chỉ số quan trọng nhất là mức chênh của giá mới nhất so với trung vị lịch sử, vì trung vị ít bị kéo lệch bởi chính các đợt giảm giá hơn giá trung bình.
def price_stats(df: pl.DataFrame) -> pl.DataFrame:
return (df.sort("day").group_by("slug", maintain_order=True)
.agg(pl.first("title"), pl.first("rating"),
pl.col("price").min().alias("min"),
pl.col("price").median().alias("median"),
pl.col("price").last().alias("latest"))
.with_columns(((pl.col("latest") / pl.col("median") - 1) * 100)
.round(1).alias("vs_median_pct")))Thứ tự sort("day") trước group_by là bắt buộc: last() chỉ có nghĩa là "giá mới nhất" khi dữ liệu đã xếp theo ngày. Bỏ sort, kết quả vẫn ra số nhưng sai, không có thông báo nào. Hàm daily_index chuẩn hoá giá trung bình mỗi ngày về mốc 100 ở ngày đầu, để một đường duy nhất cho thấy mặt bằng giá đi lên hay đi xuống:
def daily_index(df: pl.DataFrame) -> pl.DataFrame:
"""Chỉ số giá trung bình theo ngày (ngày đầu = 100)."""
d = df.group_by("day").agg(pl.col("price").mean().alias("avg")).sort("day")
return d.with_columns((pl.col("avg") / pl.col("avg").first() * 100).round(2).alias("index"))Chỉ số giá ngày cuối (2026-07-30) là 99,94 và ngày 2026-07-29 là 99,09: mặt bằng gần như đứng yên, đúng như mong đợi vì mô phỏng chỉ thêm nhiễu và giảm giá cục bộ.
Tầng 5: dashboard Streamlit
Ở cuối pipeline dữ liệu, Streamlit chạy lại toàn bộ script mỗi khi người dùng chạm vào một widget. Nếu lần nào cũng đọc DB và tính lại, trang sẽ chậm. Giải pháp là cache: theo tài liệu Streamlit về caching, st.cache_data dành cho dữ liệu tuần tự hoá được như DataFrame và trả về một bản sao mỗi lần gọi, còn st.cache_resource dành cho tài nguyên dùng chung như kết nối DB hay mô hình. Tham số ttl đặt thời gian sống của cache để dữ liệu không cũ vô hạn.
@st.cache_data(ttl=600) # đọc lại DB sau 10 phút
def load() -> pl.DataFrame:
return analyze.history()
if st.sidebar.button("Tải lại dữ liệu"):
st.cache_data.clear()
h = load()
if h.is_empty():
st.warning("Chưa có dữ liệu. Chạy run_pipeline.py trước.")
st.stop()Ba widget ở thanh bên (rating tối thiểu, ngưỡng "rẻ hơn trung vị", "chỉ hàng còn") lọc DataFrame trước khi tính chỉ số, nên thẻ số và biểu đồ luôn khớp bộ lọc. Phần thân có ba thẻ số, biểu đồ chỉ số giá, bảng sản phẩm đang rẻ và lịch sử giá của sản phẩm được chọn. Chú thích cuối trang nói thẳng rằng lịch sử ngoài ngày đầu là dữ liệu mô phỏng. Chạy bằng streamlit run app.py; khi kiểm tra, máy chủ khởi động ở chế độ headless và điểm kiểm tra sức khoẻ /_stcore/health trả về HTTP 200 với nội dung ok (lệnh và output lưu trong health_output.txt).
Streamlit hợp với dashboard của pipeline dữ liệu nội bộ và bản mẫu nhanh; khi phải phân quyền chi tiết cho khách ngoài, hãy dùng ứng dụng web có API riêng.
Kiểm thử pipeline dữ liệu
Dashboard ở cuối pipeline dữ liệu vẫn hỏng được. Streamlit có sẵn khung kiểm thử AppTest chạy ứng dụng không cần trình duyệt: AppTest.from_file("app.py").run() rồi đọc các phần tử như at.metric, at.dataframe hay at.exception, và có thể giả lập thao tác như at.slider[1].set_value(25).run(). File test_pipeline.py kiểm năm điều: parse trang 1 ra đúng 20 dòng hợp lệ, nạp lại hai lần không đổi số dòng, giá âm và trang rỗng bị validate bắt, dashboard chạy không ngoại lệ.
$ /usr/bin/time -l python test_pipeline.py
idempotent: 3000 dòng trước, 3000 dòng sau khi nạp lại 2 lần
lỗi bắt được: ['giá thiếu hoặc không dương']
trang rỗng -> ['không có dòng nào']
dashboard OK: [('Sản phẩm', '100'), ('Đang giảm giá', '2'), ('Chỉ số giá hôm nay', '99.9')]
số dòng bảng giảm giá: 2
ngưỡng 25% -> metric: [('Sản phẩm', '100'), ('Đang giảm giá', '1'), ('Chỉ số giá hôm nay', '99.9')]Dòng cuối cho thấy test thực sự tương tác với giao diện: nâng ngưỡng từ 10% lên 25% thì số sản phẩm giảm giá tụt từ 2 xuống 1, khớp với bảng ở trên (28,9% và 22,5%). Dòng "trang rỗng" cho thấy HTML đổi giao diện bị validate chặn bằng thông báo rõ ràng. Trên máy thử, bộ kiểm thử chạy vài giây, đỉnh RAM khoảng 200 MB (dòng maximum resident set size), đủ nhẹ để đặt vào CI ở mỗi lần đẩy mã (cách dựng CI xem deploy tự động bằng GitLab Runner).
Đưa vào chạy định kỳ
Một pipeline dữ liệu chỉ có giá trị khi tự chạy. Trên máy chủ Linux, một dòng cron mỗi sáng là đủ cho quy mô này; khi có nhiều job phụ thuộc nhau và cần xem lịch sử chạy, hãy dùng bộ điều phối như Jenkins hoặc Airflow. Dù chọn công cụ nào, pipeline dữ liệu cũng nên có ba thứ đi kèm:
- Tham số ngày tường minh:
crawl.py --as-of 2026-07-01cho phép chạy bù một ngày bị lỡ. - Mã thoát khác 0 khi hỏng: ngoại lệ ở
validatelàm tiến trình thoát với mã lỗi để bộ điều phối báo đỏ và gửi cảnh báo. - Đo độ mới của dữ liệu: dashboard nên hiện ngày dữ liệu mới nhất; dashboard đẹp nhưng cũ ba ngày còn nguy hiểm hơn dashboard lỗi.
Khi nào không nên dựng pipeline dữ liệu kiểu này
- Có API chính thức hoặc tệp dữ liệu (feed): pipeline dữ liệu nên đọc API ổn định hơn nhiều so với parse HTML, vốn vỡ mỗi khi trang đổi giao diện.
- Nhiều người ghi đồng thời vào pipeline dữ liệu: SQLite sẽ khoá ghi; chuyển sang PostgreSQL.
- Trang cấm crawl hoặc có dữ liệu cá nhân: dừng lại, hỏi pháp lý trước.
Lỗi thường gặp
- Tin vào dữ liệu giả lập như dữ liệu thật: luôn có cột
sourcevà chú thích trên dashboard. - Ghi bằng
INSERTthường: chạy lại là lỗi trùng khoá hoặc nhân đôi dòng. - Quên sắp xếp trước khi lấy
last(): kết quả sai mà không báo lỗi. - Cache không có
ttl: dashboard hiển thị dữ liệu cũ mãi.
Checklist
- Chia tầng cho pipeline dữ liệu rõ ràng: raw bất biến, bảng sạch có kiểm tra, phân tích tách khỏi giao diện.
- Crawl lịch sự cho pipeline dữ liệu: User-Agent nêu rõ, nghỉ giữa request, timeout, thử lại tăng dần, đọc robots.txt và điều khoản.
- Nạp idempotent trong pipeline dữ liệu: khoá chính tự nhiên, upsert, transaction; kiểm chứng bằng cách chạy hai lần.
- Dashboard có cache với ttl, nêu nguồn và ngày dữ liệu, có test chạy bằng AppTest.
Cài đặt: python3.12 -m venv .venv && .venv/bin/pip install -r requirements.txt, rồi chạy run_pipeline.py --fresh, test_pipeline.py và streamlit run app.py trong code/s7-54. Các file run_pipeline_output.txt và test_output.txt là output đã dán ở trên.
Bài tiếp theo
Dự án AI end-to-end: chatbot RAG chăm sóc khách hàng, đánh giá và triển khai. Bài này đã ghép dữ liệu thành một pipeline dữ liệu hoàn chỉnh; bài kế tiếp làm điều tương tự cho AI sinh văn bản, từ chuẩn bị tài liệu, truy hồi, đánh giá chất lượng câu trả lời tới triển khai.
