Запросы и инкрементальная обработка

Hudi поддерживает 5 типов запросов — больше, чем Delta Lake (3: snapshot, time travel, CDF) и Iceberg (4: snapshot, time travel, incremental, changelog). Это одно из ключевых преимуществ Hudi для построения инкрементальных pipeline: вместо полного пересканирования таблицы, downstream-системы читают только изменения с последнего checkpoint.

В Уроке 02 мы видели разницу между snapshot и read-optimized запросами на MOR-таблицах. Здесь мы разберём все 5 типов запросов детально, плюс паттерн checkpoint-based ETL — основу инкрементальных pipeline на Hudi.

Обзор 5 типов запросов

Snapshot Query

Snapshot query — запрос по умолчанию: возвращает полный актуальный snapshot таблицы.

COW-таблица: прямое чтение Parquet

На COW-таблице snapshot query эквивалентен чтению обычного Parquet-датасета — никакого merge overhead. Это аналогично snapshot read в Delta Lake и Iceberg.

MOR-таблица: merge base + logs

Read Optimized Query (только MOR)

Read Optimized query — уникальная для Hudi возможность: читает только base files MOR-таблицы, пропуская log files:

-- Spark SQL: выбор типа запроса
-- Snapshot (по умолчанию)
SELECT * FROM hudi_table

-- Read Optimized
SELECT * FROM hudi_table_ro
-- или через hint
SELECT * FROM hudi_table /*+ OPTIONS('hoodie.datasource.query.type'='read_optimized') */

Incremental Query

Incremental query — ключевая возможность Hudi для построения ETL-pipeline. Вместо полного пересканирования таблицы, reader получает только записи, изменённые между двумя commit timestamps:

Протокол инкрементального запроса

Incremental query работает через metadata в instant файлах: каждый completed commit содержит список затронутых файлов с их FileGroup ID. Reader:

1. Читает timeline instants в диапазоне [beginTime, endTime]
2. Извлекает список затронутых FileGroup из metadata
3. Читает только эти FileGroup (без полного scan)
4. Возвращает записи, записанные в этих commit’ах

-- Spark: incremental query
spark.read.format("hudi")
 .option("hoodie.datasource.query.type", "incremental")
 .option("hoodie.datasource.read.begin.instanttime", "20240115120000000")
 .option("hoodie.datasource.read.end.instanttime", "20240115180000000")
 .load("/path/to/hudi_table")

CDC Query

CDC query расширяет incremental query: вместо «что изменилось» возвращает как изменилось — с before/after images и op-кодами:

Включение CDC

CDC требует явного включения при создании таблицы:

# Включить CDC logging
hoodie.table.cdc.enabled=true
# Опционально: supplemental logging mode
hoodie.table.cdc.supplemental.logging.mode=data_before_after

Сравнение CDC подходов трёх форматов

Time Travel Query

Time travel позволяет прочитать snapshot таблицы на конкретный момент прошлого:

Глубина time travel

Глубина time travel определяется настройками archived timeline:

# До Hudi 1.0: archived timeline в plain Avro файлах
# Глубина ограничена hoodie.keep.min.commits (по умолчанию 20-30 instants)
# Старые instants архивируются и не всегда доступны для time travel

# Hudi 1.0+: LSM-style archived timeline
# Глубина практически неограниченна — LSM хранит compacted history
# Time travel на любой instant в истории таблицы

Checkpoint-Based ETL

Инкрементальные запросы Hudi — основа для checkpoint-based ETL: паттерна, где downstream-pipeline запоминает последний обработанный commit и при следующем запуске читает только новые изменения:

Преимущества checkpoint ETL перед full scan

Пример: multi-hop pipeline

# Этап 1: Kafka → Bronze (Hudi, MOR)
# Streaming writer каждые 30 секунд
bronze_checkpoint = "20240115120000000"

# Этап 2: Bronze → Silver (incremental ETL)
silver_df = spark.read.format("hudi") \
 .option("hoodie.datasource.query.type", "incremental") \
 .option("hoodie.datasource.read.begin.instanttime", bronze_checkpoint) \
 .load("/data/bronze/orders")

# Трансформация
silver_transformed = silver_df \
 .filter("status != 'cancelled'") \
 .withColumn("amount_usd", col("amount") * col("exchange_rate"))

# Запись в Silver (Hudi, COW для быстрого чтения)
silver_transformed.write.format("hudi") \
 .option("hoodie.table.name", "orders_silver") \
 .mode("append") \
 .save("/data/silver/orders")

# Обновить checkpoint
new_checkpoint = get_latest_commit("/data/bronze/orders")
save_checkpoint("bronze_to_silver", new_checkpoint)

hudi-rs: Python Read Path

Библиотека hudi (hudi-rs, v0.4.0) — нативный Rust-based reader для Hudi-таблиц. Поддерживает snapshot, incremental и time-travel запросы без Spark/Flink:

API примеры

import hudi

# Snapshot query
table = hudi.HudiTable("/path/to/hudi_table")
batches = table.read_snapshot() # List[RecordBatch]

# Time travel
batches = table.read_snapshot(
 timestamp="2024-01-15T12:00:00"
)

# Incremental query
batches = table.read_incremental(
 start_timestamp="2024-01-15T12:00:00",
 end_timestamp="2024-01-15T18:00:00"
)

# Конвертация в pandas
import pyarrow as pa
table_arrow = pa.Table.from_batches(batches)
df = table_arrow.to_pandas()

Дерево выбора типа запроса

Практические рекомендации

Итоги

• 5 типов запросов: snapshot, incremental, CDC, time travel, read optimized — больше, чем у Delta Lake (3) и Iceberg (4).
• Incremental query — ключевая сила Hudi: O(ΔN) вместо O(N), основа checkpoint-based ETL.
• CDC query — before/after images с op-кодами. Аналог Delta CDF, но хранится в timeline metadata, а не в отдельной директории.
• Read Optimized — уникален для MOR: быстрый, но stale. Нет аналога в Delta/Iceberg.
• Time Travel — неограниченная глубина в Hudi 1.0 (LSM archived timeline), в отличие от Delta (ограничен VACUUM retention).
• hudi-rs (v0.4.0) — Rust-based Python reader. Snapshot, incremental, time travel без Spark. Только чтение.

В следующем уроке мы разберём Table Services — async-механизмы compaction, cleaning и clustering — и экосистему интеграций Hudi с движками и форматами.