PyIceberg 0.11 у Python: повний посібник з Apache Iceberg на 2026

Оновлено: 19 червня 2026

PyIceberg, це офіційна Python-бібліотека Apache Iceberg, яка дозволяє читати, записувати та керувати таблицями Iceberg напряму з Python, без Spark, Trino чи JVM. У версії 0.11.1 (3 березня 2026) PyIceberg вже вміє читати специфікацію Iceberg v3, підтримує UPSERT, динамічні overwrite, REST-каталоги з server-side scan planning та інтегрується з PyArrow, Polars і DuckDB. Я останні пів року щодня дивлюся, як ці пайплайни тримаються о 3-й ранку, тому в цьому посібнику показую, що дійсно працює в продакшені, а що ще ні.

• Актуальна версія: PyIceberg 0.11.1 (березень 2026), вимагає Python 3.10–3.13; 3.9 більше не підтримується.
• Iceberg v3 спецификація стала GA у Snowflake (7 травня 2026) та Databricks Runtime 18, але PyIceberg підтримує v3 переважно на читання: deletion vectors і row lineage так, VARIANT, GEOMETRY та default values ні.
• Для нових проєктів обирайте REST-каталог (Apache Polaris, Lakekeeper, Unity Catalog OSS, AWS Glue Iceberg REST). Hive Metastore і чистий JDBC, це legacy.
• PyIceberg робить усі операції: DDL, DML, schema evolution, time travel, table maintenance, без Spark-кластера.
• Основні болі продакшену: small files problem, manifest bloat, відсутність авто-компакції; план компакції потрібно мати з першого дня.
• Інтеграції з PyArrow, Polars, DuckDB та pandas через to_arrow(), to_polars(), to_duckdb() працюють з predicate pushdown і scan planning на стороні сервера.

Що таке PyIceberg і навіщо він потрібен у 2026

Apache Iceberg, це відкритий формат таблиць для аналітичних data lakes: він додає до файлів Parquet/ORC шар метаданих, що дає ACID-транзакції, snapshot isolation, schema evolution і time travel. Історично, щоб писати в Iceberg, потрібен був Spark, Flink або Trino. PyIceberg усуває цей бар'єр: він пише та читає Iceberg напряму з Python, не запускаючи JVM. У моїй практиці це означає, що CDC-пайплайн з Postgres у lakehouse тепер виглядає як 80 рядків Python із psycopg та pyiceberg.table.upsert(), а не цілий Spark Structured Streaming-джоб з власним кластером.

Версія 0.11 додала server-side scan planning у REST-каталогах: клієнт надсилає запит на сканування, а каталог повертає вже розпланований список файлів. Це знімає з Python-процесу 90% роботи з метаданими для великих таблиць. Окремо в 0.11 додали підтримку читання ORC через PyArrow і Iceberg v3 deletion vectors. Якщо у вас pipeline на dbt-Python-моделях або Dagster-asset'ах, які матеріалізують Iceberg-таблиці, оновлення з 0.10 → 0.11 дає видимий приріст продуктивності навіть без зміни коду.

Встановлення PyIceberg 0.11 та залежності

PyIceberg розповсюджується як pyiceberg на PyPI з опціональними «extras» для кожного каталогу та FileIO. Мінімально достатній набір для роботи з REST-каталогом та S3:

python -m venv .venv
source .venv/bin/activate

# REST + S3 + PyArrow + pandas
pip install "pyiceberg[s3fs,pyarrow,pandas]==0.11.1"

# Для AWS Glue REST endpoint
pip install "pyiceberg[glue,s3fs,pyarrow]==0.11.1" boto3

# Якщо хочете Polars-інтеграцію
pip install "pyiceberg[pyarrow]==0.11.1" "polars>=1.20"

З 0.11 офіційно підтримуються Python 3.10–3.13. Python 3.9 видалили в 0.10, а коли я перевіряла на 3.13 із PyArrow 18.1, усе працювало без жодних варнінгів. Окремо існує пакет pyiceberg-core (0.9.1, травень 2026), це Rust-бекенд на базі iceberg-rust, який поступово замінює частини Python-реалізації. Для більшості користувачів він ставиться автоматично як транзитивна залежність.

Каталоги: REST, Glue, Polaris, Nessie, SQL

Iceberg, це формат, що крутиться навколо каталогу. Каталог знає, де лежить актуальний metadata.json кожної таблиці, і виконує atomic swap при коміті. Який каталог обрати в 2026?

Каталог	Тип	PyIceberg підтримка	Коли обирати
Apache Polaris	REST, OSS	Read + Write	Vendor-neutral, нова інсталяція; став TLP у Apache 18 лютого 2026
Lakekeeper	REST, Rust	Read + Write	Single binary, без JVM, k8s-deployment
AWS Glue Iceberg REST	Managed REST	Read + Write	Все вже в AWS, S3 Tables федеруються через s3tablescatalog
Unity Catalog OSS	REST	Read + Write	Команди на Databricks, потрібна сумісність з Delta
Project Nessie	REST + Git-like	Read + Write	Multi-table atomic transactions та гілкування
SQL Catalog (Postgres/SQLite)	JDBC	Read + Write	Локальна розробка, тести
Hive Metastore	Thrift	Read + Write (legacy)	Існуючі Hadoop-кластери, плануйте міграцію

Мій дефолтний вибір: REST-каталог. Один клієнт обслуговує Polaris, Lakekeeper, Glue, Unity Catalog. Конфіг для REST-каталогу в ~/.pyiceberg.yaml:

catalog:
  prod:
    type: rest
    uri: https://polaris.internal.example.com/api/catalog
    warehouse: my_warehouse
    credential: "client_id:client_secret"
    scope: "PRINCIPAL_ROLE:ALL"

Або напряму в коді (корисно для тестів та notebook'ів):

from pyiceberg.catalog import load_catalog

catalog = load_catalog(
    "prod",
    **{
        "type": "rest",
        "uri": "https://polaris.internal.example.com/api/catalog",
        "warehouse": "my_warehouse",
        "credential": "client_id:client_secret",
        "scope": "PRINCIPAL_ROLE:ALL",
    },
)
print(catalog.list_namespaces())

Для Glue зміни type: rest на type: glue та додайте glue.region. Якщо ви використовуєте AWS S3 Tables, найкраще не йти через нативний Glue-клієнт, а через Glue Iceberg REST endpoint, він SigV4-підписаний і дає credential vending з коробки. Детальніше про вибір каталогу див. на офіційній сторінці конфігурації PyIceberg.

Як створити Iceberg-таблицю в Python

Щоб створити таблицю, потрібні три речі: namespace (база даних у термінах Iceberg), schema (опис колонок) і опційно partition spec. Ось мінімальний приклад для подієвого пайплайну з CDC:

from pyiceberg.catalog import load_catalog
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

catalog = load_catalog("prod")
catalog.create_namespace_if_not_exists("analytics")

schema = Schema(
    NestedField(1, "event_id", LongType(), required=True),
    NestedField(2, "user_id", LongType(), required=True),
    NestedField(3, "event_type", StringType(), required=True),
    NestedField(4, "amount", DoubleType()),
    NestedField(5, "occurred_at", TimestampType(), required=True),
)

partition_spec = PartitionSpec(
    PartitionField(
        source_id=5, field_id=1000,
        transform=DayTransform(), name="day",
    )
)

table = catalog.create_table(
    identifier=("analytics", "events"),
    schema=schema,
    partition_spec=partition_spec,
    properties={
        "format-version": "2",
        "write.parquet.compression-codec": "zstd",
        "write.target-file-size-bytes": "134217728",  # 128 MB
    },
)

Кілька деталей, які я б хотіла знати, коли починала. По-перше, у партиціонуванні я завжди вказую field_id явно (починаючи з 1000), бо це робить діагностику легшою (якщо потім читати table.spec() через інший engine, ім'я і id будуть стабільними. По-друге, format-version: "2", це поточний дефолт, але якщо ваш downstream engine ще не вміє v2 deletion files, лишайте v2 явно та не переходьте поки на v3 у property. По-третє, target-file-size-bytes у 128 МБ, це компроміс для більшості аналітичних запитів; для high-write workloads я ставлю 256 МБ, щоб компактити рідше.

Запис, UPSERT та динамічні overwrite

Записати pandas DataFrame у щойно створену таблицю можна одним рядком:

import pyarrow as pa
import pandas as pd

df = pd.DataFrame({
    "event_id": [1, 2, 3],
    "user_id": [100, 101, 100],
    "event_type": ["purchase", "view", "purchase"],
    "amount": [29.90, None, 14.50],
    "occurred_at": pd.to_datetime([
        "2026-06-18 10:01:00",
        "2026-06-18 10:05:00",
        "2026-06-19 09:00:00",
    ]),
})

table.append(pa.Table.from_pandas(df))

Для CDC-сценаріїв, де приходить апдейт існуючого рядка, у 0.10 додали справжній upsert(), який матчить за заданими join keys і робить merge-on-read через positional deletes:

updates = pa.Table.from_pylist([
    {"event_id": 1, "user_id": 100, "event_type": "purchase",
     "amount": 31.90, "occurred_at": pd.Timestamp("2026-06-18 10:01:00")},
    {"event_id": 4, "user_id": 102, "event_type": "view",
     "amount": None, "occurred_at": pd.Timestamp("2026-06-19 11:30:00")},
])

result = table.upsert(updates, join_cols=["event_id"])
print(result.rows_updated, result.rows_inserted)

Для batch-перезапису одного дня без переписування решти партицій є dynamic_overwrite() з 0.10:

from pyiceberg.expressions import EqualTo
from datetime import date

day_data = pa.Table.from_pandas(today_df)
table.dynamic_partition_overwrite(day_data)

Schema evolution без переписування таблиці

Іceberg трекає колонки за стабільним числовим field_id, тому додавання, видалення та перейменування колонок, це лише зміна метаданих, без переписування жодного Parquet-файлу. У PyIceberg API оформлений як контекстний менеджер транзакції:

with table.update_schema() as update:
    update.add_column("currency", StringType(),
                      doc="ISO 4217 currency code", required=False)
    update.rename_column("amount", "amount_usd")
    update.update_column("amount_usd", field_type=DoubleType())
    update.delete_column("event_type")

Що важливо для дата-інженера: Iceberg заборонить зміну, яка зруйнує сумісність зі снапшотами (наприклад, drop required column без default). Це частина схема-валідації у транзакції, і її не обійти випадково. Time travel запити на старі снапшоти продовжать працювати, якщо ви не перейменували колонку у фільтрі. Подивіться роботу з ширшим pipeline-контекстом у нашому посібнику з автоматизованого очищення даних, де я показую, як зв'язати schema evolution з pandera-контрактами.

Time travel та керування снапшотами

Кожен коміт у Iceberg створює snapshot, незмінне посилання на повний стан таблиці. PyIceberg дозволяє читати конкретний snapshot за id або за wall-clock timestamp:

# Подивитися всі snapshot'и
for snap in table.snapshots():
    print(snap.snapshot_id, snap.timestamp_ms, snap.summary["operation"])

# Прочитати стан 6 годин тому
import time
six_hours_ago_ms = int((time.time() - 6 * 3600) * 1000)
old_snapshot = next(
    s for s in reversed(table.snapshots())
    if s.timestamp_ms <= six_hours_ago_ms
)

old_data = table.scan(snapshot_id=old_snapshot.snapshot_id).to_arrow()

Якщо потрібно відкотити продуктивний коміт, який зламав downstream-моделі dbt, є rollback_to_snapshot(). Я тримаю в Dagster-asset'і sentinel-перевірку: якщо row count впав > 30% у порівнянні з минулою матеріалізацією, asset падає та автоматично робить manage_snapshots().rollback_to(parent_id).commit(). Це той тип «pipeline test», який економить дзвінок о 3-й ранку.

Снапшоти не безкоштовні: їх метадані ростуть лінійно з кількістю комітів. Експіріть старе через expire_snapshots хоча б щодня. Дефолтний retention становить 5 днів історії, що логічно для CDC, але мало для аудиту фінансових даних.

Приховане партиціонування та partition transforms

Унікальна фіча Iceberg, це hidden partitioning. Користувач пише WHERE occurred_at = '2026-06-19', а Iceberg під капотом мапить це на день-партицію без жодних додаткових date_trunc у запиті. PyIceberg підтримує всі основні partition transforms:

• IdentityTransform, пряме партиціонування за значенням
• YearTransform / MonthTransform / DayTransform / HourTransform
• BucketTransform(n), хешований bucket з фіксованою кількістю
• TruncateTransform(n), обрізання строки/числа до n

Partition spec можна змінювати на існуючій таблиці без перезапису:

from pyiceberg.transforms import HourTransform

with table.update_spec() as update:
    update.add_field(
        source_column_name="occurred_at",
        transform=HourTransform(),
        partition_field_name="hour",
    )

Старі дані лишаються розбиті по днях, нові пишуться по годинах. Iceberg уміє читати обидва набори в одному запиті. Це partition evolution, і саме за неї варто прийти з Hive в Iceberg, навіть якщо ви не використовуєте нічого з v3.

Інтеграція з PyArrow, Polars і DuckDB

PyIceberg повертає скани як PyArrow Table, тому будь-який downstream engine, що говорить Arrow, працює з коробки. Найкорисніші три варіанти:

scan = table.scan(
    row_filter="event_type = 'purchase' AND occurred_at >= '2026-06-01'",
    selected_fields=("event_id", "user_id", "amount_usd"),
)

# Як Arrow Table, для подальшої обробки в PyArrow
arrow_tbl = scan.to_arrow()

# Як Polars LazyFrame, predicate pushdown зберігається
lf = scan.to_polars()
result = lf.group_by("user_id").agg(pl.sum("amount_usd")).collect()

# Як DuckDB relation, SQL-аналітика без копіювання даних
import duckdb
con = duckdb.connect()
duck_rel = scan.to_duckdb(table_name="events", connection=con)
con.sql("SELECT user_id, SUM(amount_usd) FROM events GROUP BY 1").show()

Predicate з row_filter просочується вниз до Iceberg-планувальника, який відсіче нерелевантні файли через статистику min/max у metadata. Це partition pruning + file skipping безкоштовно. Якщо ви хочете глибше зрозуміти Arrow-шар, я писала окремий посібник з Apache Arrow та PyArrow 18; а для аналітичних запитів, які виходять за межі однієї таблиці, краще зчитати у DuckDB і робити join'и там. Polars-сценарії описані в практичному посібнику з Polars.

Чи підтримує PyIceberg специфікацію Iceberg v3

Коротко: частково, переважно на читання. Iceberg v3 спецификація стала GA у Snowflake 7 травня 2026 і в Databricks Runtime 18. Сім ключових можливостей v3: deletion vectors (Puffin-based, ~10× швидші DML), row lineage (_row_id, _last_updated_sequence_number для native CDC), VARIANT semi-structured type, GEOMETRY/GEOGRAPHY з CRS, default column values, nanosecond timestamps, multi-argument partition transforms.

Стан у PyIceberg 0.11 (червень 2026):

• Deletion vectors: читання так (з 0.10, покращено в 0.11), запис ні.
• Row lineage: читання так (0.11), запис ні.
• Nanosecond timestamps: читання так (0.10), запис частково.
• VARIANT, GEOMETRY/GEOGRAPHY, default values, multi-arg partition transforms: ні або експериментально.

Якщо ваш Spark/Snowflake уже пише v3, PyIceberg їх прочитає. Якщо ви хочете самі писати v3 з Python, поки що рано: ставте format-version: "2" на створенні таблиці. Деталі по релізах слідкуйте на сторінці релізів iceberg-python на GitHub.

Продакшен-чекліст: компакція, тестування, моніторинг

Я говорила це команді безліч разів: Iceberg-пайплайн без компакції і без тестів, це таймер до інциденту. Ось мінімальний продакшен-чекліст, який я вимагаю від кожної нової матеріалізації:

1. Compaction job щодня. PyIceberg ще не має повноцінного асинхронного compaction (на відміну від Spark/Trino через CALL rewrite_data_files). Для більшості lakehouse я раджу запустити Spark job або скористатися авто-компакцією S3 Tables. Без цього у вас за тиждень буде 50 000 файлів по 1 МБ і Trino падатиме на планувальнику.
2. Expire snapshots щодня. Викликайте table.expire_snapshots().expire_older_than(retention_ms).commit(). Дефолтні 5 днів підходять для CDC; для аудиту тримайте 30–90, але робіть це усвідомлено.
3. Row count test після кожної матеріалізації. Якщо це dbt-Python модель, додайте not_null, unique та власний row_count_change_within_range у YAML. Якщо Dagster, asset_check з sentinel'ом ±30%.
4. Schema contract з pandera. Перед append() валідуйте Arrow-таблицю проти схеми, інакше колонка з NaN там, де очікувався int64, тихо перетвориться на null-string у Parquet.
5. Catalog-level access control. Polaris і Unity Catalog OSS мають role-based ACL; не пишіть DML з принципала, який має доступ до прод-каталогу без обмежень.
6. Monitor manifest count і file count. Експортуйте table.inspect.manifests() та table.inspect.files() у вашу observability, це найдешевший сигнал «треба компактити».

PyIceberg чи Spark для Iceberg?

Питання, яке я чую найчастіше. Відповідь залежить від обсягу даних і вже існуючої інфраструктури. PyIceberg виграє, коли таблиці меншi за петабайти, є один читач/писар на партицію, і у вас немає бажання тримати JVM-кластер. Spark виграє, коли треба паралельний distributed compute, регулярна важка компакція, або коли ваші writes природно йдуть зі Spark Structured Streaming.

На практиці я бачу гібрид: PyIceberg робить інжест (CDC, REST API → Iceberg), а планові job'и компакції, expire snapshots і великі бекфіли виконує Spark раз на добу. Так ви маєте легкий Python-pipeline для оперативних задач і важку артилерію тільки тоді, коли вона дійсно потрібна. Для архітектурного огляду див. сторінку Apache Polaris, який створено саме для того, щоб обидва ці шляхи бачили однаковий каталог.

Часті запитання

Чи готовий PyIceberg до продакшену в 2026 році?

Так, для більшості OLAP-навантажень. PyIceberg 0.11.1 стабільно підтримує DDL, DML, time travel і schema evolution. Головні обмеження продакшену: відсутність вбудованої компакції (плануйте окремий Spark або S3 Tables auto-maintenance) і неповний write-шлях для Iceberg v3.

Який каталог Iceberg обрати для нового проєкту?

За замовчуванням REST-каталог. Конкретний вибір: AWS Glue Iceberg REST endpoint, якщо ви в AWS; Apache Polaris (TLP з лютого 2026) для vendor-neutral self-hosted; Lakekeeper для Rust-фірст інсталяцій на Kubernetes; Unity Catalog OSS, якщо вам важлива сумісність із Delta Lake.

Як зробити UPSERT з Python в Iceberg-таблицю?

З PyIceberg 0.10 використовуйте table.upsert(arrow_table, join_cols=["id"]). Метод матчить вхідні рядки за заданими join-колонками і виконує merge-on-read через positional delete files. Для повного перезапису однієї партиції краще dynamic_partition_overwrite().

Чи можна виконувати time travel у PyIceberg?

Так. Викликайте table.scan(snapshot_id=...).to_arrow() для конкретного снапшоту або шукайте по wall-clock timestamp у table.snapshots(). Для відкату використовуйте table.manage_snapshots().rollback_to(snapshot_id).commit(). Не забудьте про expire_snapshots, інакше історія росте безконтрольно.

Чи підтримує PyIceberg специфікацію Iceberg v3?

Частково. PyIceberg 0.11 читає v3 deletion vectors, row lineage та nanosecond timestamps, але не пише v3 і не підтримує типи VARIANT, GEOMETRY/GEOGRAPHY, default column values та multi-argument partition transforms. Якщо ваш write-шлях, це PyIceberg, лишайтесь на format-version: "2".