Исходная задача и контекст

Перед нами стояла типичная для дата‑инженеров задача: обработать поток Parquet‑файлов с данными о внутренних технических процессах заказчика. Ключевой запрос — извлечь метаданные из таблицы, чтобы в дальнейшем работать с ними быстрее и удобнее.

В качестве основного инструмента мы выбрали Apache Iceberg — проверенный в наших проектах фреймворк для управления метаданными. Изначально пайплайн выглядел так:

  • Парсинг и сохранение метаданных в Postgres БД.

  • Выгрузка файлов в память (порциями размером row_group).

  • Разделение файлов на «чанки».

  • Добавление колонки file_name для идентификации источника.

  • Запись и регистрация файлов в Iceberg.

  • Пометка файлов как прочитанных.

На первый взгляд — стандартная схема. Но при масштабировании вскрылись серьёзные проблемы.

Проблемы первой реализации

Когда мы перешли к параллельным загрузкам, Iceberg начал выдавать ошибки. В документации предлагалось решение через retry (optimistic concurrency) — повторные попытки записи при конфликтах. Однако в Python‑библиотеке (PyIceberg) эта функциональность отсутствовала в нужной форме(время обработки не соответствовало ожиданиям из-за того, что система тратила ресурсы на излишние операции загрузки данных в S3 при ошибках) и нам пришлось реализовывать её самостоятельно.

Поиск решения: разделение операций

Мы решили перестроить процесс, разделив два этапа:

  • Загрузка данных (в нашем случае через PyArrow).

  • Обновление метаданных (через PyIceberg).

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

  • Файлы отдельно записываются в S3.

  • PyIceberg занимается исключительно регистрацией изменений в метаданных: проверяет уникальность, обновляет ссылки, но не трогает сами файлы. 

from pyiceberg.io.pyarrow import parquet_files_to_data_files

.
.
.


table = catalog.load_table((self.namespace, table_name))

data_file = next(
	parquet_files_to_data_files(
	    io=table.io,
	    file_paths=iter([file_path]),
	    table_metadata=table.metadata,
	)
)

.
.
.

@retry(
    wait=wait_random_exponential(multiplier=0.5, max=2),
    stop=stop_after_attempt(20),
    reraise=True,
    before_sleep=_log_retry,
)
def _add_data_file_with_retry(
    self, table: IcebergTable, data_file: DataFile
) -> None:
	table.refresh()
    with table.transaction() as transaction:
        transaction.update_snapshot().fast_append().append_data_file(
            data_file
        ).commit()

_add_data_file_with_retry(table=table, data_file=data_file)

Отметим, при создании таблицы и добавлении данных pyiceberg формируются две логические «папки»:

  • metadata/ — хранит метаинформацию (схемы, партиции, версии);

  • data/ — содержит файлы в S3, но файлы могут находиться в любом месте S3‑бакета (даже в другом бакете). Iceberg лишь фиксирует их пути в метаданных. Например:

«Файл data_part_1.parquet лежит по пути s3://bucket/raw/2024/, и в нём содержатся данные за январь 2024 года».

Это даёт гибкость: можно не перемещать файлы внутри S3  — достаточно обновить метаданные.

Новая схема: минимум операций, максимум скорости

Мы отказались от громоздкой схемы с выгрузкой/загрузкой и перешли к упрощённому пайплайну:

  • Парсинг и сохранение метаданных в Postgres БД (как и раньше).

  • Регистрация файлов в Iceberg с указанием их текущих путей в S3.

Что изменилось:

  • Исчезла необходимость выгружать файлы из S3 для обработки.

  • Iceberg анализирует метаданные напрямую в бакете, без перемещения данных.

  • Операции выполняются за доли секунды — основное время уходит на межсервисное взаимодействие.

После внедрения новой схемы:

  • Скорость обработки выросла в разы: операция, занимавшая минуты, теперь выполняется за полсекунды.

  • Нагрузка на инфраструктуру упала: нет избыточных перемещений данных, память и сеть не перегружены.

  • Гибкость системы повысилась: файлы можно не перемещать внутри S3, а Iceberg автоматически актуализирует ссылки.

  • Масштабируемость улучшилась: система готова к росту объёмов без срочного апгрейда железа.

Этот кейс показывает, что иногда решение проблемы лежит не в наращивании ресурсов, а в пересмотре архитектуры. Связка Iceberg + S3 позволила:

  • сократить время обработки данных;

  • снизить затраты на инфраструктуру;

  • оставить задел для будущего роста.

Теперь система работает быстрее, потребляет меньше ресурсов и остаётся понятной для поддержки.