Vấn đề thực tế: Pipeline ETL chạy xong, dữ liệu gốc biến mất
Mình từng gặp tình huống này: pipeline ETL chạy vào lúc 2 giờ sáng, transform dữ liệu sales, ghi đè lên file Parquet cũ. Sáng ra team analytics phàn nàn kết quả báo cáo tháng trước bỗng dưng khác. Lục lại log không tìm ra nguyên nhân vì dữ liệu gốc đã bị overwrite hoàn toàn.
Chuyện này không phải trường hợp cá biệt. Với pipeline quy mô vừa, hễ job ghi đè Parquet là mất trắng lịch sử snapshot — không có cách nào roll back mà không tốn công backup thủ công trước. Nhiều team giải quyết bằng cách đặt tên file theo ngày: sales_2024_11_01.parquet, sales_2024_11_02.parquet… nhưng cách này tốn storage, query phức tạp, và vẫn không có ACID guarantees.
Tại sao Parquet thuần túy không đủ?
Parquet rất tốt cho analytics workload — columnar storage, nén tốt, đọc nhanh. Nhưng Parquet là file format thuần túy, không có khái niệm transaction hay versioning. Khi ghi đè một file Parquet:
- Không có lịch sử thay đổi (audit trail)
- Không rollback được nếu job lỗi giữa chừng
- Schema evolution rất khó — thêm/xóa cột dễ làm hỏng pipeline downstream
- Concurrent write từ nhiều job dẫn đến race condition, data corruption
Parquet không có lỗi gì ở đây — nó làm đúng việc được thiết kế. Cái thiếu là một lớp metadata management và transaction ngay phía trên. Đây cũng là hệ quả điển hình khi thiết kế schema database không tính đến vòng đời dữ liệu từ sớm.
Apache Iceberg + DuckDB giải quyết như thế nào
Iceberg ra đời để lấp đúng chỗ trống đó. Bên dưới vẫn là Parquet, nhưng Iceberg thêm một lớp metadata management với bốn tính năng cốt lõi:
- Snapshot-based versioning: Mỗi lần write tạo snapshot mới, không xóa snapshot cũ
- ACID transactions: Không lo concurrent write corruption
- Schema evolution: Thêm/xóa/rename cột mà không cần rewrite toàn bộ data
- Time Travel: Query dữ liệu tại bất kỳ thời điểm nào trong quá khứ
DuckDB đóng vai trò query engine: nhẹ, nhanh, chạy ngay trên laptop. Từ version 0.10+, extension iceberg chính thức hỗ trợ đọc Iceberg tables trực tiếp. Toàn bộ stack local — không cần AWS, không cần Hive Metastore.
Cài đặt môi trường
Chỉ cần Python 3.9+, không phụ thuộc service nào khác.
# Tạo virtual environment
python -m venv iceberg-env
source iceberg-env/bin/activate
# Cài PyIceberg và DuckDB
pip install "pyiceberg[pyarrow,duckdb]" duckdb pandas pyarrowKiểm tra version sau khi cài:
import pyiceberg
import duckdb
print(pyiceberg.__version__) # >= 0.7.0
print(duckdb.__version__) # >= 0.10.0Tạo Iceberg Catalog và Table đầu tiên
PyIceberg cần một catalog để quản lý metadata. Với môi trường cục bộ, dùng SqlCatalog lưu metadata vào SQLite — không cần setup bất kỳ service nào thêm.
from pyiceberg.catalog.sql import SqlCatalog
import pyarrow as pa
import os
os.makedirs("warehouse/data", exist_ok=True)
catalog = SqlCatalog(
"local",
**{
"uri": "sqlite:///warehouse/catalog.db",
"warehouse": "file://warehouse",
}
)
catalog.create_namespace("analytics")
schema = pa.schema([
pa.field("order_id", pa.int64()),
pa.field("customer_id", pa.int64()),
pa.field("amount", pa.float64()),
pa.field("status", pa.string()),
pa.field("created_at", pa.timestamp("us")),
])
table = catalog.create_table("analytics.orders", schema=schema)
print("Table created:", table.identifier)Ghi dữ liệu và tạo Snapshots
import pandas as pd
from datetime import datetime, timedelta
# Batch 1: Dữ liệu ngày hôm qua
df1 = pd.DataFrame({
"order_id": [1, 2, 3],
"customer_id": [101, 102, 103],
"amount": [150.0, 230.5, 89.99],
"status": ["completed", "pending", "completed"],
"created_at": [datetime.now() - timedelta(days=1)] * 3
})
table.append(pa.Table.from_pandas(df1))
snapshot_1 = table.current_snapshot().snapshot_id
print(f"Snapshot 1 ID: {snapshot_1}")
# Batch 2: Thêm đơn mới
df2 = pd.DataFrame({
"order_id": [4, 5],
"customer_id": [104, 105],
"amount": [320.0, 45.0],
"status": ["completed", "refunded"],
"created_at": [datetime.now()] * 2
})
table.append(pa.Table.from_pandas(df2))
snapshot_2 = table.current_snapshot().snapshot_id
print(f"Snapshot 2 ID: {snapshot_2}")Lúc này Iceberg giữ cả 2 snapshots. Dữ liệu gốc của snapshot 1 vẫn còn nguyên trên disk, không bị xóa hay overwrite — đây chính là điều Parquet thuần túy không làm được.
Time Travel Query với DuckDB
Giờ đến phần đáng giá nhất. DuckDB đọc trực tiếp Iceberg table, kể cả query theo snapshot cụ thể:
import duckdb
conn = duckdb.connect()
conn.execute("INSTALL iceberg; LOAD iceberg;")
# Query trạng thái hiện tại — đủ 5 đơn hàng
result = conn.execute("""
SELECT * FROM iceberg_scan('warehouse/analytics/orders')
ORDER BY order_id
""").fetchdf()
print("Current state (5 rows):")
print(result)
# Time Travel — query snapshot 1 (chỉ 3 đơn đầu tiên)
result_v1 = conn.execute(f"""
SELECT * FROM iceberg_scan(
'warehouse/analytics/orders',
snapshot_id = {snapshot_1}
)
ORDER BY order_id
""").fetchdf()
print(f"\nSnapshot {snapshot_1} — 3 rows only:")
print(result_v1)Nếu pipeline lúc 2 giờ sáng ghi đè nhầm, mình query lại snapshot trước đó là ra ngay dữ liệu gốc — không cần restore backup, không cần downtime.
Schema Evolution — Thêm cột không phá vỡ pipeline cũ
from pyiceberg.types import FloatType
# Thêm cột discount_pct vào schema
with table.update_schema() as update:
update.add_column("discount_pct", FloatType())
# Các snapshot cũ vẫn đọc được — cột mới trả NULL cho dữ liệu cũ
result = conn.execute("""
SELECT order_id, amount, discount_pct
FROM iceberg_scan('warehouse/analytics/orders')
""").fetchdf()
print(result)
# discount_pct = NaN cho 5 đơn cũ — hoàn toàn hợp lệVới Parquet thuần túy, thêm cột như vậy thường làm hỏng reader của snapshot cũ vì schema không khớp. Iceberg xử lý tự động — cột mới trả NULL cho dữ liệu cũ, reader cũ không bị break.
Tips và Best Practices từ thực chiến
1. Đặt retention policy cho snapshots ngay từ đầu
Snapshots tích lũy theo thời gian, tốn disk. Nên định kỳ expire:
from datetime import datetime, timedelta
# Giữ lại snapshot trong 7 ngày gần nhất
table.expire_snapshots(
older_than_ms=int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
).commit()
print("Expired old snapshots")2. Partition bảng theo thời gian để tăng tốc query
Khai báo partition ngay khi tạo table (không thêm vào sau được với Iceberg v1):
from pyiceberg.partitioning import PartitionSpec, PartitionField
from pyiceberg.transforms import DayTransform
# source_id=5 là index của cột created_at trong schema
spec = PartitionSpec(
PartitionField(
source_id=5,
field_id=1000,
transform=DayTransform(),
name="created_day"
)
)
# Khai báo khi tạo table
table = catalog.create_table(
"analytics.orders_partitioned",
schema=schema,
partition_spec=spec
)DuckDB sẽ chỉ scan đúng partition cần thiết thay vì đọc toàn bộ file. Với bảng 100GB, query chỉ cho một ngày cụ thể có thể nhanh hơn 20–50x so với full scan.
3. Chuẩn hóa dữ liệu nguồn trước khi ingest
Khi nhận dữ liệu từ nhiều nguồn, mình hay gặp trường hợp cần convert CSV sang JSON để chuẩn hóa trước khi đẩy vào Iceberg. Với file nhỏ dưới 10MB, mình dùng converter trên trình duyệt tại toolcraft.app/vi/tools/data/csv-to-json — chạy client-side nên không sợ lộ data nhạy cảm, tiện hơn nhiều so với viết script one-off cho từng nguồn.
4. Kết hợp với dbt cho transform phức tạp
DuckDB có adapter chính thức cho dbt (dbt-duckdb). Toàn bộ stack hoàn toàn local: Iceberg storage + DuckDB compute + dbt transforms = mini data warehouse không tốn một xu cloud. Phù hợp cho team nhỏ hoặc proof-of-concept trước khi scale lên production.
Cách tốt nhất để bắt đầu
Đang quản lý data pipeline với Parquet files và đã từng mất dữ liệu hoặc không trace được lịch sử? Mình khuyên migrate sang Iceberg theo từng bước:
- Bắt đầu với bảng ít critical nhất, test Time Travel trên môi trường dev
- Dùng
SqlCatalogvới SQLite cho môi trường cục bộ — không cần infra phức tạp - Set
expire_snapshotspolicy từ đầu, không để sau mới nghĩ đến - Sau khi quen, migrate lên REST Catalog hoặc AWS Glue cho production nếu cần
Iceberg không phải công cụ phù hợp cho mọi trường hợp — với bảng nhỏ dưới vài triệu rows, overhead metadata management không đáng công. Nhưng khi data lên đến vài chục GB, hoặc khi audit trail là bắt buộc (tài chính, compliance), đây là lựa chọn tốt nhất mình từng dùng. Sau khi chuyển sang stack này, mình không còn thức đêm debug pipeline nữa.
