Hidden partitioning и partition evolution

В предыдущих уроках мы разобрали иерархию метаданных и механику снимков. Теперь — одна из самых сильных сторон Iceberg: hidden partitioning (скрытое партиционирование) и partition evolution (эволюция партиционирования).

В Hive и Delta Lake пользователь явно указывает partition-колонки, и структура партиций видна в файловой системе. Iceberg полностью скрывает партиционирование от пользователя — запросы пишутся по оригинальным колонкам, а движок автоматически применяет partition pruning.

Проблема явного партиционирования

В Hive и Delta Lake партиционирование — явное: пользователь создаёт partition-колонку, данные раскладываются в директории, и запросы должны явно фильтровать по partition-колонке.

-- Hive / Delta Lake: явное партиционирование
CREATE TABLE orders (
 order_id BIGINT,
 amount DECIMAL(10,2),
 order_ts TIMESTAMP
) PARTITIONED BY (order_date DATE); -- пользователь создал колонку

-- Запрос ДОЛЖЕН использовать partition-колонку
SELECT * FROM orders WHERE order_date = '2025-01-15'; -- OK, partition pruning
SELECT * FROM orders WHERE order_ts >= '2025-01-15'; -- FULL SCAN! Движок не знает связь order_ts → order_date

Проблемы:

Hidden partitioning — решение Iceberg

В Iceberg партиционирование скрыто от пользователя. Вместо partition-колонок используются transforms — функции, применяемые к существующим колонкам:

Ключевые преимущества:

1. Запросы по оригинальным колонкам — нет производных partition-колонок
2. Автоматический pruning — движок знает связь между фильтром и partition transform
3. Нет дублирования данных — partition values хранятся в метаданных (manifest file), не в данных
4. Partition evolution — можно изменить партиционирование без перезаписи данных

Partition transforms

Iceberg определяет набор transforms — функций, применяемых к source-колонке для вычисления partition value:

Примеры использования transforms

from pyiceberg.catalog import load_catalog
from pyiceberg.transforms import (
 IdentityTransform,
 YearTransform,
 MonthTransform,
 DayTransform,
 HourTransform,
 BucketTransform,
 TruncateTransform,
)
from pyiceberg.table.partitioning import PartitionSpec, PartitionField
import pyarrow as pa

catalog = load_catalog("demo", type="sql",
 uri="sqlite:////tmp/iceberg_demo/catalog.db",
 warehouse="file:///tmp/iceberg_demo")

catalog.create_namespace_if_not_exists("analytics")

schema = pa.schema([
 ("event_id", pa.int64()),
 ("user_id", pa.int64()),
 ("event_ts", pa.timestamp("us", tz="UTC")),
 ("event_type", pa.string()),
 ("payload", pa.string()),
])

# Партиционирование: day(event_ts) + bucket[16](user_id)
# Движок автоматически вычисляет partition values из данных
table = catalog.create_table(
 "analytics.events",
 schema=schema,
 partition_spec=PartitionSpec(
 PartitionField(
 source_id=3, # event_ts (field id)
 field_id=1000,
 transform=DayTransform(),
 name="event_ts_day"
 ),
 PartitionField(
 source_id=2, # user_id (field id)
 field_id=1001,
 transform=BucketTransform(num_buckets=16),
 name="user_id_bucket"
 ),
 ),
)

print(f"Partition spec: {table.spec()}")

Как transforms работают с запросами

Для каждого transform Iceberg знает projection — как преобразовать предикат на source-колонке в предикат на partition value:

Per-file partition statistics

В manifest file каждая запись data_file содержит partition — struct со значениями partition полей для этого файла. Это позволяет двухуровневый pruning:

Partition statistics metadata table

Iceberg 1.10.x поддерживает incremental partition stats — агрегированная статистика по partition values, доступная через metadata table partition_stats:

-- Spark: просмотр partition statistics
SELECT partition, record_count, file_count, total_data_file_size_in_bytes
FROM db.events.partitions;

# pyiceberg: просмотр через plan_files
scan = table.scan()
partition_stats = {}
for task in scan.plan_files():
 pv = str(task.file.partition)
 if pv not in partition_stats:
 partition_stats[pv] = {"files": 0, "records": 0, "bytes": 0}
 partition_stats[pv]["files"] += 1
 partition_stats[pv]["records"] += task.file.record_count
 partition_stats[pv]["bytes"] += task.file.file_size_in_bytes

for pv, stats in sorted(partition_stats.items()):
 print(f" {pv}: {stats['files']} files, "
 f"{stats['records']} records, "
 f"{stats['bytes'] / 1024 / 1024:.1f} MB")

Partition evolution

Partition evolution — возможность изменить партиционирование без перезаписи данных. Это уникальная особенность Iceberg, отсутствующая в Delta Lake и Hive.

Как это работает

Каждый partition spec имеет spec-id. Manifest files ссылаются на partition-spec-id. При partition evolution:

1. Создаётся новый partition spec с новым spec-id
2. Новый spec становится default-spec-id в metadata file
3. Новые data files записываются с новым partition spec
4. Старые data files остаются как есть — их manifest files ссылаются на старый spec

Partition evolution через pyiceberg

# Текущий partition spec
print(f"Current spec: {table.spec()}")
# PartitionSpec(day(3, 'event_ts_day', 1000), bucket[16](2, 'user_id_bucket', 1001))

# Эволюция: добавить hour вместо day
with table.update_spec() as update:
 update.remove_field("event_ts_day") # убрать day
 update.add_field(
 source_column_name="event_ts",
 transform=HourTransform(),
 name="event_ts_hour"
 )

print(f"New spec: {table.spec()}")
# PartitionSpec(void(3, 'event_ts_day', 1000), bucket[16](2, 'user_id_bucket', 1001), hour(3, 'event_ts_hour', 1002))

# Обратите внимание: day transform заменён на void (а не удалён),
# чтобы старые manifest files оставались валидными

Типичные сценарии partition evolution

Multi-spec read path

При чтении Iceberg обрабатывает manifest files с разными partition specs прозрачно для пользователя:

Оптимизация старых данных (опционально)

После partition evolution можно опционально перезаписать старые файлы с новым partition spec для улучшения pruning:

# Spark: перезапись старых файлов с новым partition spec
# CALL system.rewrite_data_files(
# table => 'db.events',
# where => 'event_ts < "2025-03-01"'
# )
#
# Это НЕ обязательно — таблица работает корректно с обоими specs.
# Перезапись — оптимизация для улучшения pruning старых данных.

Projection и residual predicates

Не все предикаты полностью покрываются partition pruning. Оставшаяся часть — residual predicate — применяется при чтении data files:

Запрос: WHERE event_ts >= '2025-03-15 10:30:00'
Partition spec: day(event_ts)

Partition predicate: day >= 20166 (15 марта) → pruning
Residual predicate: event_ts >= '2025-03-15 10:30:00' → фильтр при чтении

Partition pruning пропускает файлы за 14 марта и раньше. Но файл с day=20166 (15 марта) содержит записи за весь день — с 00:00 до 23:59. Residual predicate >= 10:30 фильтрует строки при чтении этого файла.

# pyiceberg автоматически применяет row_filter
scan = table.scan(
 row_filter="event_ts >= '2025-03-15T10:30:00+00:00'"
)

# Можно посмотреть planned tasks
for task in scan.plan_files():
 print(f" File: {task.file.file_path}")
 print(f" Partition: {task.file.partition}")
 print(f" Records: {task.file.record_count}")
 # Residual filter применяется при to_arrow()

result = scan.to_arrow()
print(f"Rows after residual filter: {len(result)}")

Составные partition specs

Iceberg поддерживает несколько partition fields в одном spec:

# Пример: таблица событий с составным партиционированием
spec = PartitionSpec(
 PartitionField(
 source_id=3, # event_ts
 field_id=1000,
 transform=DayTransform(),
 name="event_ts_day"
 ),
 PartitionField(
 source_id=2, # user_id
 field_id=1001,
 transform=BucketTransform(num_buckets=16),
 name="user_id_bucket"
 ),
 PartitionField(
 source_id=4, # event_type
 field_id=1002,
 transform=IdentityTransform(),
 name="event_type"
 ),
)

При запросе с фильтром по нескольким колонкам Iceberg применяет pruning по всем partition fields одновременно:

SELECT * FROM events
WHERE event_ts >= '2025-03-15' -- pruning по day
 AND user_id = 42 -- pruning по bucket
 AND event_type = 'purchase' -- pruning по identity

Движок вычисляет: day >= 20166 AND bucket = 6 AND event_type = 'purchase', и читает только файлы, совпадающие по всем трём partition полям.

Выбор partition strategy

Рекомендации по выбору transforms:

# Пример: выбор bucket count
# Правило: N × средний размер партиции ≈ target file size (256-512 MB)
# Если таблица ~50 GB и target file size = 512 MB:
# 50 GB / 512 MB ≈ 100 файлов → bucket[16] даёт ~6 файлов на бакет
# Слишком мало файлов на бакет → увеличить до bucket[32] или bucket[64]

Сравнение с Delta Lake и Hive

Итоги

1. Hidden partitioning — пользователь не видит партиции; запросы по оригинальным колонкам, pruning автоматический
2. 8 transforms — identity, year, month, day, hour, bucket, truncate, void — покрывают все сценарии
3. Двухуровневый pruning — manifest list (partition range summary) → manifest file (per-file partition values)
4. Partition evolution — мгновенное изменение гранулярности (month→day, day→hour) без перезаписи данных
5. Multi-spec read — старые файлы со старым spec + новые файлы с новым spec читаются прозрачно
6. void transform — корректное отключение партиционирования по полю при эволюции
7. Residual predicates — часть фильтра, не покрытая partition pruning, применяется при чтении data files