Iceberg Sink и Confluent Tableflow

В 2025 году streaming и lakehouse окончательно сошлись в одной архитектуре. Kafka хранит события несколько часов или дней (retention), Iceberg/Delta Lake — годами. Между ними нужен мост: компонент, который материализует поток сообщений в колоночные файлы object-storage и регистрирует их в каталоге как таблицу. Этот мост строится двумя способами: open-source Apache Iceberg Sink Connector для Kafka Connect и управляемый сервис Confluent Tableflow.

Этот урок объясняет оба подхода, их commit-семантику, schema evolution и production-tradeoffs.

Контекст: Streaming-Lakehouse convergence

Классическая архитектура разделяла операционный поток (Kafka) и аналитическое хранилище (Snowflake, BigQuery, Spark). Между ними стояли batch ETL-задачи: каждый час Spark-job читал Kafka, дедуплицировал, писал в Parquet. Лаг — часы. Дублирование данных — двойное хранилище.

В 2024–2025 годах переход на open table formats (Apache Iceberg, Delta Lake, Apache Hudi) изменил картину. Iceberg-таблица — это не просто Parquet-файлы, а транзакционная абстракция со snapshot-isolation и атомарными commit. Это позволило построить streaming ingestion: коннектор пишет файлы прямо в object-storage и фиксирует snapshot, как только пакет готов.

Apache Iceberg Sink Connector

IcebergSinkConnector — официальный open-source sink, разработанный командой Tabular (acquired Databricks в 2024) и теперь являющийся частью Apache Iceberg core. Доступен в apache/iceberg репозитории, поставляется как plugin для Kafka Connect.

Ключевые свойства:

• Exactly-once семантика через KIP-447 (Kafka 2.5+) — данные пишутся транзакционно: data files + offset commit в одной Kafka-транзакции.
• Multi-table routing — один коннектор может писать в несколько Iceberg-таблиц (fan-out), маршрутизация по полю или по топику.
• Schema evolution — автоматическое добавление новых столбцов на основе схемы из Schema Registry.
• CDC mode — обработка Debezium-событий с upsert/delete вместо append-only.
• REST/Glue/Hive/Nessie/JDBC catalog — поддержка всех основных каталогов.

Минимальная конфигурация

{
  "name": "iceberg-events-sink",
  "config": {
    "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector",
    "tasks.max": "4",

    "topics": "user-events",
    "iceberg.tables": "analytics.user_events",

    "iceberg.catalog.type": "rest",
    "iceberg.catalog.uri": "https://catalog.example.com",
    "iceberg.catalog.warehouse": "s3://my-warehouse/",
    "iceberg.catalog.io-impl": "org.apache.iceberg.aws.s3.S3FileIO",

    "iceberg.tables.auto-create-enabled": "true",
    "iceberg.tables.evolve-schema-enabled": "true",
    "iceberg.control.commit.interval-ms": "300000",

    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

Параметры разбираются по группам:

Commit strategy и exactly-once

В отличие от обычного sink connector, который коммитит offset после каждого put(), Iceberg sink использует двухэтапный коммит координатора и worker-задач.

Алгоритм exactly-once:

1. Coordinator раз в commit.interval-ms рассылает begin-commit event в control topic.
2. Каждый worker получает событие, закрывает текущий Parquet writer и отправляет data-files event со списком файлов и Kafka-offset.
3. Coordinator собирает все data-files events и формирует один Iceberg snapshot атомарно. В тот же момент Kafka offset коммитится транзакционно (KIP-447).
4. Если crash случился до коммита — workers удалят оркфаненные файлы при перезапуске и переписали данные. Если после — Iceberg snapshot уже виден читателям, offset закоммичен.

Schema evolution

Если iceberg.tables.evolve-schema-enabled=true, коннектор автоматически расширяет схему таблицы при появлении новых полей в сообщениях.

Multi-table routing и CDC

Один коннектор может писать в несколько Iceberg-таблиц. Два режима:

Static routing — список таблиц в iceberg.tables, маршрутизация по топику:

"topics": "orders,payments,refunds",
"iceberg.tables": "warehouse.orders,warehouse.payments,warehouse.refunds"

Dynamic routing — таблица определяется значением поля в сообщении:

"iceberg.tables.dynamic-enabled": "true",
"iceberg.tables.route-field": "table_name"

При dynamic routing каждое сообщение содержит поле table_name, и коннектор маршрутизирует в соответствующую таблицу. Это полезно для Debezium CDC: одна Debezium-source пишет события из всех таблиц БД в один топик с метаданными, а Iceberg sink раскладывает их по таблицам lakehouse.

CDC mode

Для Debezium-событий включается iceberg.tables.cdc-field:

"iceberg.tables.cdc-field": "op",
"iceberg.tables.upsert-mode-enabled": "true"

Коннектор интерпретирует op (insert / update / delete) и применяет equality deletes + new data files. Iceberg merge-on-read формат сохраняет delete-маркеры, которые применяются при чтении.

Confluent Tableflow

В 2025 году Confluent объявил GA своего управляемого аналога — Tableflow. Концепция: каждый Kafka topic в Confluent Cloud можно одной кнопкой превратить в Iceberg или Delta Lake таблицу.

Tableflow GA включает:

• Iceberg + Delta Lake одновременно — один топик может быть представлен в обоих форматах параллельно.
• Catalog integration — автопубликация в AWS Glue, Snowflake Open Catalog (Polaris), Databricks Unity Catalog.
• Built-in maintenance — автоматическая компакция мелких файлов, snapshot expiration, оркфаненные файлы.
• Dead Letter Queue — schema-violation сообщения изолируются без остановки потока.
• Bring Your Own Key — customer-managed encryption keys для compliance.

Архитектура Tableflow

В отличие от Iceberg sink connector, Tableflow не использует Kafka consumer API. Сервис читает сегменты напрямую из tiered storage Confluent Cloud.

Tableflow читает segment-файлы напрямую из tiered storage, минуя consumer API. Это даёт большую пропускную способность и не нагружает брокеров. Компромисс — это работает только в Confluent Cloud (vendor lock-in).

Сравнение подходов

В production выбор сводится к четырём вариантам:

Limitations и production tips

Limitations Iceberg Sink Connector

1. Commit interval = latency floor. Меньше 1 минуты ставить нельзя без серьёзных проблем с производительностью. 5 минут — здоровый default.
2. Append-only по умолчанию. Equality deletes требуют CDC mode и дополнительной нагрузки на reader.
3. Maintenance не входит в коннектор. Нужен отдельный job (Spark, Trino, Amoro) для компакции, snapshot expiration, оркфаненных файлов.
4. Рост числа small files. При высоком commit-rate каждый snapshot создаёт N файлов (по числу partitions × workers). Без компакции query performance деградирует.

Production tips

{
  "iceberg.control.commit.interval-ms": "300000",
  "iceberg.tables.default-partition-by": "days(event_time)",
  "iceberg.tables.default-properties.write.target-file-size-bytes": "536870912",
  "iceberg.tables.default-properties.write.distribution-mode": "hash"
}

Maintenance jobs

Запустите отдельные Spark/Trino задачи на расписании:

-- Компакция мелких файлов
CALL system.rewrite_data_files('analytics.user_events',
  options => map('target-file-size-bytes', '536870912'));

-- Удаление старых snapshot (старше 7 дней)
CALL system.expire_snapshots('analytics.user_events',
  older_than => TIMESTAMP '2025-04-23 00:00:00');

-- Удаление оркфаненных файлов
CALL system.remove_orphan_files(table => 'analytics.user_events');

Когда какой подход

Ключевые выводы

1. Streaming-Lakehouse pattern в 2025 — стандарт: Kafka topic эквивалентен Iceberg table через коннектор или managed-сервис.
2. Apache Iceberg Sink Connector — open-source путь, exactly-once через KIP-447, commit-interval (default 5 минут) задаёт минимальный latency.
3. Confluent Tableflow — управляемый сервис в Confluent Cloud, читает segment-файлы напрямую, секунды latency, встроенный maintenance, Iceberg + Delta Lake одновременно.
4. Schema evolution автоматическая в обоих подходах: новые столбцы добавляются на metadata-уровне, без переписывания файлов.
5. CDC mode в Iceberg sink через cdc-field + upsert + equality deletes; в Tableflow CDC поддерживается из коробки.
6. Production maintenance обязателен: компакция файлов, snapshot expiration, orphan files. Tableflow делает это сам, для open-source connector нужен отдельный job.
7. REST catalog (Polaris, Lakekeeper, Snowflake Open Catalog, Unity) — стандарт для streaming-lakehouse в 2025; статические Hive/JDBC каталоги уходят.