PyIceberg 2026: Python Iceberg Không Sp…

Cập nhật: 28 tháng 7, 2026

PyIceberg là thư viện Python thuần túy cho phép đọc và ghi bảng Apache Iceberg mà không cần JVM, Spark hay Hive Metastore. Trong bản 0.9 phát hành đầu năm 2026, PyIceberg hỗ trợ đầy đủ ghi dữ liệu (append, overwrite, delete), quản lý snapshot, schema evolution và tích hợp trực tiếp với pandas, Polars, PyArrow và DuckDB. Bài viết này hướng dẫn triển khai một pipeline lakehouse hoàn chỉnh bằng PyIceberg, kèm ví dụ code chạy được, so sánh với Delta-rs và các mẹo tôi rút ra khi vận hành pipeline xử lý hàng chục tỷ event mỗi ngày.

PyIceberg 0.9 (tháng 3/2026) là bản đầu tiên hỗ trợ ghi đầy đủ: append, overwrite, delete và merge-on-read đều chạy thuần Python trên PyArrow.
Không cần Spark hay JVM. Bạn có thể xây dựng một lakehouse hoàn chỉnh bằng PyIceberg cộng với object storage (S3/GCS/MinIO) và REST catalog (Polaris, Lakekeeper, Nessie).
Đọc bảng Iceberg dưới dạng pandas.DataFrame, polars.DataFrame hoặc DuckDB relation chỉ với một dòng code sau khi load catalog.
PyIceberg hỗ trợ time travel qua snapshot_id, schema evolution an toàn (add/rename/drop cột) và partition spec động, ba tính năng quan trọng nhất khiến Iceberg hơn Parquet trần.
So với Delta-rs, PyIceberg thắng ở đa engine (Trino, Snowflake, BigQuery đều đọc được) nhưng hiện chậm hơn khi ghi bảng phân vùng dày đặc.
Dùng chunked write và commit theo batch để tránh bùng nổ manifest (vấn đề tôi gặp trong tháng đầu triển khai production).

PyIceberg là gì và tại sao dùng vào năm 2026?

PyIceberg là bản triển khai Python thuần túy của đặc tả Apache Iceberg, một định dạng bảng mở (open table format) đặt trên các file Parquet trong object storage. Trước năm 2024, muốn ghi Iceberg gần như bắt buộc phải chạy Spark hoặc Trino; muốn đọc thì có nhiều lựa chọn hơn nhưng vẫn cần JVM. Bản 0.9 phát hành ngày 12 tháng 3 năm 2026 đã đóng nốt khoảng trống cuối cùng khi hỗ trợ đầy đủ delete và merge-on-read, đưa PyIceberg thành client Iceberg đầy đủ tính năng đầu tiên không phụ thuộc JVM.

Trong thực tế, tôi đã ngừng đưa Spark vào các pipeline batch nhỏ dưới 500 GB/giờ và chuyển sang PyIceberg từ quý IV năm 2025. Lý do rất đơn giản: một worker Python chạy trên máy 8 vCPU với PyArrow 20 và PyIceberg 0.9 xử lý xong ingestion trong 4 phút, trong khi cùng khối lượng đó chạy Spark on Kubernetes tốn 3 phút warm-up cluster trước khi thực sự bắt đầu. Với các job dưới ngưỡng "big data thật sự", overhead của JVM chiếm nhiều hơn công việc thực. PyIceberg cắt gọn stack đến mức tối thiểu: một tiến trình Python, một object store, một REST catalog.

Các thành phần chính bạn cần biết: catalog (nơi lưu metadata bảng, ví dụ Polaris, Nessie, Lakekeeper, AWS Glue, hoặc REST catalog tự viết), table metadata (file JSON mô tả schema, partition spec, snapshot hiện tại), manifest list (danh sách các manifest thuộc một snapshot), và data files (Parquet thật sự). PyIceberg quản lý toàn bộ ba tầng metadata này để bạn chỉ cần suy nghĩ theo mô hình bảng.

Cài đặt PyIceberg và kết nối REST catalog

PyIceberg được đóng gói theo kiểu "core mỏng, extras nhiều". Bạn chỉ cài phần thực sự dùng để tránh kéo về SDK AWS, Google Cloud hay Azure không cần thiết. Cài đặt tối thiểu cho pipeline S3 cộng với REST catalog:

pip install "pyiceberg[s3fs,pyarrow,duckdb,polars]==0.9.0"
pip install "pandas==2.3.0" "pyarrow==20.0.0"

Cấu hình catalog bằng file ~/.pyiceberg.yaml hoặc bằng biến môi trường. Ví dụ với REST catalog (Lakekeeper, Polaris hoặc Nessie 0.104 đều dùng chuẩn REST catalog spec):

catalog:
  prod:
    type: rest
    uri: https://catalog.internal.example.com
    warehouse: s3://lakehouse-prod/warehouse
    credential: ${LAKEKEEPER_TOKEN}
    s3.endpoint: https://s3.eu-west-1.amazonaws.com
    s3.region: eu-west-1

Sau đó load catalog trong Python:

from pyiceberg.catalog import load_catalog

catalog = load_catalog("prod")
print(catalog.list_namespaces())
# [('analytics',), ('marketing',), ('raw',)]

Nếu bạn muốn thử nhanh không cần server, PyIceberg đi kèm SQLite catalog dùng cho local dev, cực kỳ tiện cho unit test:

from pyiceberg.catalog.sql import SqlCatalog

catalog = SqlCatalog(
    "local",
    **{
        "uri": "sqlite:///:memory:",
        "warehouse": "file:///tmp/warehouse",
    },
)
catalog.create_namespace("demo")

Trong CI tôi hầu như luôn dùng SqlCatalog thay vì mock. Nó chạy đúng code path thật, chỉ khác backend metadata. Đây cũng là chiến lược tôi khuyến nghị cho các bài test ETL nói chung, tương tự cách tiếp cận trong bài validation dữ liệu bằng Pandera cho pipeline production: test với dữ liệu thật, đừng mock schema.

Đọc Iceberg table trong pandas, Polars và DuckDB

Đọc là phần mượt nhất của PyIceberg. Sau khi có tham chiếu bảng, bạn chuyển sang engine mình quen dùng chỉ với một lời gọi hàm. PyIceberg tự động push down predicate (filter theo cột phân vùng) và column projection xuống Parquet reader, nên bạn không mất performance khi đi qua tầng abstraction này.

from pyiceberg.expressions import GreaterThanOrEqual

table = catalog.load_table("analytics.orders")

# Lấy pandas DataFrame với filter push-down và chỉ 3 cột
df = (
    table.scan(
        row_filter=GreaterThanOrEqual("order_date", "2026-07-01"),
        selected_fields=("order_id", "customer_id", "total_cents"),
    )
    .to_pandas()
)
print(df.dtypes)

Với Polars (engine nhanh nhất khi làm việc trên DataFrame lớn hơn RAM), chỉ cần đổi hậu tố:

import polars as pl

lazy = table.scan(row_filter=GreaterThanOrEqual("order_date", "2026-07-01")).to_polars()
result = lazy.filter(pl.col("total_cents") > 10_000).group_by("customer_id").agg(pl.len())
print(result)

Bạn có thể tham khảo thêm về sức mạnh của lazy evaluation trong bài hướng dẫn Polars DataFrame siêu tốc. Kết hợp với PyIceberg là combo tôi dùng cho hầu hết job ad-hoc analytics hiện nay.

Cách yêu thích của tôi là dùng DuckDB. Bạn chạy SQL ANSI trực tiếp trên Iceberg mà không cần export sang file trung gian:

duck = table.scan().to_duckdb(table_name="orders")
result = duck.sql("""
    SELECT customer_id, SUM(total_cents) / 100.0 AS gmv_usd
    FROM orders
    WHERE order_date >= DATE '2026-07-01'
    GROUP BY customer_id
    ORDER BY gmv_usd DESC
    LIMIT 20
""").df()

Nếu bạn chưa quen với DuckDB, bài hướng dẫn DuckDB Python 2026 là điểm khởi đầu tốt. Kết hợp DuckDB với PyIceberg cho phép bạn chạy interactive analytics trên hàng tỷ dòng ngay trong một notebook mà không cần warehouse thương mại.

Ghi dữ liệu vào Iceberg từ Python (không cần Spark)

Đây là phần thay đổi lớn nhất trong bản 0.9. Trước đây bạn chỉ append được; giờ overwrite, delete và upsert đều hoạt động ổn định trên PyArrow tables. Tạo bảng mới:

import pyarrow as pa
from pyiceberg.schema import Schema
from pyiceberg.types import (
    NestedField, LongType, StringType, TimestampType, DoubleType
)
from pyiceberg.partitioning import PartitionSpec, PartitionField
from pyiceberg.transforms import DayTransform

schema = Schema(
    NestedField(1, "order_id", LongType(), required=True),
    NestedField(2, "customer_id", LongType(), required=True),
    NestedField(3, "order_date", TimestampType(), required=True),
    NestedField(4, "product_sku", StringType(), required=False),
    NestedField(5, "total_cents", LongType(), required=True),
    NestedField(6, "usd_rate", DoubleType(), required=False),
)

partition_spec = PartitionSpec(
    PartitionField(source_id=3, field_id=1000, transform=DayTransform(), name="order_day")
)

table = catalog.create_table(
    "analytics.orders",
    schema=schema,
    partition_spec=partition_spec,
    properties={"write.parquet.compression-codec": "zstd"},
)

Ghi append từ PyArrow (đây cũng là con đường bạn dùng cho pandas hay Polars, cả hai đều chuyển sang Arrow gần như không tốn phí):

arrow_batch = pa.Table.from_pylist([
    {"order_id": 1, "customer_id": 42, "order_date": "2026-07-28T09:00:00",
     "product_sku": "SKU-A", "total_cents": 15900, "usd_rate": 1.08},
    {"order_id": 2, "customer_id": 91, "order_date": "2026-07-28T09:01:00",
     "product_sku": "SKU-B", "total_cents": 4200, "usd_rate": 1.08},
], schema=table.schema().as_arrow())

table.append(arrow_batch)

Overwrite theo filter (idempotent, an toàn để retry trong Airflow/Prefect):

from pyiceberg.expressions import EqualTo

table.overwrite(
    df=arrow_batch,
    overwrite_filter=EqualTo("order_day", "2026-07-28"),
)

Xóa theo predicate (thường dùng cho GDPR "right to be forgotten"):

table.delete(delete_filter=EqualTo("customer_id", 91))

Delete tạo ra positional delete files (theo spec Iceberg v2/v3) thay vì rewrite toàn bộ data files, nên rất rẻ. Sau này bạn có thể chạy rewrite_data_files để compact. Đây là bước bảo trì quan trọng khi bảng có nhiều delete tích lũy.

Nếu bạn đang xây dựng pipeline ELT end-to-end, phần ghi này ghép rất tự nhiên với dlt cho ELT pipeline declarative. dlt lo phần source, PyIceberg lo phần sink, và bạn giữ toàn bộ pipeline trong Python.

Schema evolution, partition spec và time travel

Ba tính năng này là lý do người ta chọn Iceberg thay vì Parquet trần trong data lake. PyIceberg 0.9 hỗ trợ tất cả qua transaction API:

from pyiceberg.types import BooleanType

with table.update_schema() as update:
    update.add_column("is_refunded", BooleanType(), required=False)
    update.rename_column("usd_rate", "fx_rate_usd")
    update.delete_column("product_sku")

Iceberg đảm bảo backward-compatible reads: các reader cũ vẫn đọc được dữ liệu vì mỗi cột được định danh bằng field_id chứ không phải tên. Đây là điểm quan trọng phân biệt Iceberg với Hive-style tables. Bạn có thể rename cột mà không phá pipeline nào.

Đổi partition spec (evolution) mà không cần rewrite dữ liệu cũ:

from pyiceberg.transforms import MonthTransform

with table.update_spec() as update:
    update.remove_field("order_day")
    update.add_field("order_date", MonthTransform(), "order_month")

Iceberg lưu partition spec theo từng snapshot, nên dữ liệu ghi trước đây vẫn tổ chức theo order_day, còn dữ liệu mới đi vào order_month. Reader tự động ghép hai partition strategy khi scan.

Time travel qua snapshot_id hoặc timestamp:

# Xem lịch sử snapshot
for snap in table.snapshots():
    print(snap.snapshot_id, snap.timestamp_ms, snap.summary)

# Đọc bảng ở trạng thái tại một snapshot cụ thể
old = table.scan(snapshot_id=snap.snapshot_id).to_pandas()

# Hoặc theo timestamp Unix (ms)
history = table.scan(as_of_timestamp=1_753_600_000_000).to_pandas()

Với các pipeline chạy hàng đêm, tôi luôn ghi lại current_snapshot_id vào bảng metadata riêng để có thể roll back trong vòng vài giây nếu phát hiện dữ liệu upstream sai. Tính năng này gần như miễn phí về mặt lưu trữ vì Iceberg chỉ giữ metadata, còn data files được reuse cross-snapshot.

PyIceberg vs Delta-rs vs Hudi Python: nên chọn cái nào?

Ba định dạng lakehouse mở phổ biến nhất năm 2026 đều đã có client Python thuần túy. Đây là ma trận so sánh dựa trên trải nghiệm sản xuất và benchmark nội bộ tôi chạy hồi tháng 6:

Tiêu chí	PyIceberg 0.9	delta-rs 0.24	PyHudi 0.2 (alpha)
Cần JVM/Spark	Không	Không	Có (một phần)
Ghi (append/overwrite/delete)	Đầy đủ	Đầy đủ	Chỉ append
Multi-engine reader	Trino, Snowflake, BigQuery, DuckDB, Databricks	Databricks, Trino, DuckDB	Trino, Spark
Time travel	Có	Có	Có
Catalog options	REST, Glue, Hive, SQL, in-memory	Unity, filesystem	Timeline server
Tốc độ ghi bảng phân vùng dày	Trung bình	Nhanh nhất	Chậm
Ecosystem 2026	Đang tăng nhanh nhất	Ổn định	Đang chuyển tiếp

Nếu team bạn đã chạy Databricks toàn bộ, delta-rs là lựa chọn tự nhiên vì tương thích trực tiếp với Unity Catalog. Nếu bạn có nhiều engine truy vấn (ví dụ Snowflake cho BI, DuckDB cho ad-hoc, Trino cho federation), Iceberg thắng cách biệt về khả năng đọc chéo. Hudi rất mạnh cho streaming upsert nhưng client Python vẫn ở giai đoạn alpha, và tôi chưa đưa vào bất kỳ pipeline production nào.

Với việc AWS S3 Tables và Google BigLake đều chọn Iceberg làm định dạng mặc định trong năm 2025 và đặc tả Iceberg V3 được phê duyệt cuối 2025, momentum đang nghiêng rõ về phía Iceberg cho các dự án mới.

Best practices khi vận hành PyIceberg trong production

Đây là những bài học đắt tiền tôi rút ra sau chín tháng chạy PyIceberg trên pipeline khoảng 120 tỷ event/ngày. Áp dụng chúng ngay từ đầu để tránh phải rewrite bảng sau này.

1. Commit theo batch, không commit từng row

Mỗi lời gọi append() tạo ra một snapshot mới với metadata file riêng. Nếu bạn commit 10 000 lần mỗi giờ, sau một tuần bảng có 1.7 triệu snapshot và mỗi lần list-partitions mất vài phút. Gom dữ liệu vào PyArrow table đủ lớn (thường 128 đến 512 MB) rồi commit một lần. Tôi ăn cú vấp này ngay tuần đầu chạy prod, phải viết script gộp snapshot cả buổi tối.

2. Chạy expire_snapshots hàng ngày

Iceberg không xóa snapshot cũ tự động. Đặt job hằng ngày:

from datetime import datetime, timedelta, timezone

cutoff = datetime.now(timezone.utc) - timedelta(days=7)
table.expire_snapshots(expire_older_than=int(cutoff.timestamp() * 1000)).commit()

3. Rewrite data files sau chiến dịch delete lớn

Sau khi delete GDPR, tỉ lệ delete files/data files có thể vượt ngưỡng khiến reader chậm. Chạy compaction định kỳ để merge delete files vào data files mới.

4. Đặt properties đúng ngay lúc tạo bảng

Ba property tôi luôn set: write.parquet.compression-codec=zstd (nhỏ hơn snappy khoảng 20%), write.target-file-size-bytes=134217728 (128 MB, tránh small file problem), và write.metadata.delete-after-commit.enabled=true.

5. Orchestrate bằng Prefect hoặc Airflow, không cron

Vì mỗi commit là transaction, retry an toàn là yêu cầu cứng. Tôi khuyến nghị dùng Prefect 3 với async flows cho orchestration để orchestrate. Bạn được retry tự động, observability và có thể trigger PyIceberg jobs từ event streams mà không tốn overhead Airflow scheduler.

6. Log snapshot_id vào metadata pipeline

Với mỗi job ghi thành công, lưu lại table.current_snapshot().snapshot_id vào bảng audit. Đây là bảo hiểm ba đồng cho một triệu: khi upstream data bị lỗi, bạn chỉ cần manage_snapshots().rollback_to_snapshot(id).commit() là quay về trạng thái sạch. Honestly, tính năng này đã cứu tôi ít nhất hai lần trong sáu tháng gần đây.

Câu hỏi thường gặp

PyIceberg có thay thế được Spark không?

Có, cho hầu hết các workload batch dưới 1 TB/giờ. Với các job cần shuffle nặng (join nhiều bảng lớn) hoặc streaming true realtime, Spark hoặc Flink vẫn phù hợp hơn. Nhưng cho ingestion, transformation vừa phải và maintenance job, PyIceberg 0.9 đủ mạnh và tiết kiệm khoảng 60% chi phí infra so với chạy Spark on Kubernetes.

PyIceberg có hỗ trợ merge (upsert) không?

Chưa có API merge cấp cao trong 0.9, nhưng bạn có thể mô phỏng bằng delete() theo khóa rồi append() dữ liệu mới trong cùng một transaction. Tính năng merge_rows() đang được phát triển và dự kiến ra mắt trong 1.0 vào cuối năm 2026 theo roadmap chính thức của dự án.

Có thể dùng PyIceberg với Snowflake hoặc BigQuery không?

Được. Snowflake hỗ trợ Iceberg tables với REST catalog từ 2024 và BigQuery hỗ trợ Iceberg qua BigLake. Bạn dùng PyIceberg để ghi, còn Snowflake/BigQuery đọc trực tiếp cùng bảng đó. Đây chính là ưu điểm lớn nhất của open table format so với format độc quyền.

Cần catalog nào để chạy PyIceberg trong production?

REST catalog là lựa chọn hiện đại nhất. Lakekeeper (open source, Rust), Apache Polaris (đã tốt nghiệp incubator tháng 2/2026) và Nessie đều hoạt động tốt với PyIceberg. AWS Glue Catalog phù hợp nếu bạn đã ở trong AWS. Tránh Hadoop filesystem catalog cho production vì không đảm bảo atomic commit khi ghi song song.

PyIceberg đọc được bảng do Spark ghi và ngược lại không?

Có, hoàn toàn tương thích ở mức spec. Đây chính là điểm mạnh của Iceberg: bảng do Spark, Flink, Trino hay PyIceberg ghi đều là cùng một định dạng chuẩn. Bạn có thể migration dần từ Spark sang PyIceberg mà không phải dừng pipeline cũ, chính là thứ tôi đã làm với ba pipeline production trong sáu tháng qua.