Всем привет! Меня зовут Александр Андреев, я ведущий инженер данных в департаменте машинного обучения компании "АльфаСтрахование". Я люблю изучать новые и перспективные технологии в сфере обработки и хранения данных, а еще больше я люблю рассказывать о них коллегам и внедрять их в рабочие процессы. В этой статье я хочу сделать обзор не совсем новой, но при этом перспективной опенсорс-технологии хранения данных - Apache Paimon. Мы пройдемся детально от возникновения потребности в streamhouse-подходе к хранению данных и основ Apache Paimon до сравнительных бенчмарков с другими подходами к хранению данных и примеров кода. Возможно, именно эта технология подойдет вашей компании для того, чтобы наконец "поженить" батч со стримингом.

Введение: эволюция хранения данных и текущие вызовы

Давайте представим современную data-платформу крупной компании. С одной стороны, у вас есть системы, генерирующие непрерывный поток событий: клики пользователей, транзакции, логи сервисов. С другой стороны, аналитики и дата-сайентисты ждут свежие данные для построения отчетов и обучения моделей. Между системами и людьми традиционно стоит сложная инфраструктура из множества компонентов: Kafka для потоковой передачи, Flink или Spark Streaming для обработки, HDFS или S3 для хранения, и наконец, хранилище данных вроде ClickHouse или Snowflake для аналитики.

Каждый переход между компонентами добавляет задержку. Каждая система требует своего формата данных. Каждое преобразование может привести к потере консистентности. В результате получается то, что в индустрии называют "data swamp" - болото данных, где простой вопрос "какие данные у нас актуальные?" превращается в настоящий детектив.

Apache Paimon появился как попытка решить эту фундаментальную проблему, объединив лучшее из двух подходов: эффективность потоковой обработки и надежность батч-систем. Но чтобы понять, почему это важно, давайте сначала разберемся с корнем проблемы.

Корень проблемы: почему streaming и batch живут раздельно

Исторически сложилось так, что системы для обработки потоков и пакетной обработки развивались независимо, решая разные задачи. Потоковые системы оптимизировались для минимальной задержки - получил событие, обработал, отправил дальше. Их главная метрика успеха - это latency (задержка), то есть время от появления данных до получения результата. Пакетные системы, напротив, оптимизировались для максимальной пропускной способности - собрал большой объем данных, эффективно обработал все разом. Их метрика - throughput (пропускная способность), то есть количество данных, обработанных за единицу времени.

Эти различия привели к совершенно разным архитектурным решениям. Потоковые системы хранят данные в append-only логах, где новые записи просто добавляются в конец. Это быстро для записи, но неэффективно для аналитических запросов. Пакетные системы используют колоночное хранение и индексы, что отлично для аналитики, но требует дорогостоящей реорганизации данных при каждой записи.

Попытки объединить эти подходы обычно заканчивались компромиссами: либо вы жертвовали скоростью записи ради эффективности чтения, либо наоборот. Лямбда-архитектура, популярная в 2010-х, предлагала поддерживать две параллельные системы - speed layer для потоков и batch layer для исторических данных. Однако это удваивало сложность и стоимость инфраструктуры, а главное - не гарантировало консистентность между слоями.

Что такое Apache Paimon

Apache Paimon решает эту дилемму элегантным способом, используя структуру данных LSM-tree (Log-Structured Merge-tree) в сочетании с современной lakehouse архитектурой. Давайте разберем каждый компонент по отдельности, а затем посмотрим, как они работают вместе.

Начнем с LSM-tree: это структура данных, которая была изобретена еще в 1996 году, но только сейчас нашла свое идеальное применение в контексте больших данных. Основная идея проста: вместо того чтобы сразу записывать данные в оптимизированную для чтения структуру, мы сначала быстро записываем их в неупорядоченный лог, а затем в фоновом режиме переорганизовываем для эффективного чтения.

Представим это как работу с заметками. Когда вам нужно быстро что-то записать, вы хватаете первый попавшийся листок и пишете. Это быстро, но найти потом нужную информацию сложно. Поэтому периодически вы садитесь и переписываете заметки в организованный блокнот (или в популярный у it-специалистов инструмент Obsidian), группируя по темам, добавляя оглавление. LSM-tree работает по тому же принципу: быстрая запись в "черновик" (Level 0), периодическая реорганизация в "чистовик" (Level 1, 2, ...).

Lakehouse архитектура добавляет к этому второй важный элемент - унификацию метаданных и поддержку ACID транзакций. Традиционный data lake - это просто набор файлов в распределенной файловой системе. У вас есть данные, но нет гарантий консистентности, нет схемы, нет возможности делать атомарные обновления. Lakehouse добавляет слой метаданных, который превращает набор файлов в полноценную таблицу с транзакциями, версионированием и оптимизацией запросов.

Paimon объединяет эти концепции, создавая систему, где данные физически хранятся в LSM-tree структуре, но логически представляются как таблицы с полной поддержкой SQL-операций. При этом - и это ключевой момент - одни и те же данные доступны как для потокового чтения (следим за новыми изменениями), так и для пакетных запросов (анализируем исторические данные).

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

Архитектура Apache Paimon состоит из нескольких ключевых компонентов, каждый из которых решает свою часть общей задачи.

В основе лежит многоуровневая система хранения файлов. Когда данные поступают в Paimon, они сначала накапливаются в памяти в специальной структуре, называемой MemTable (прямо как в ScyllaDB). Это позволяет группировать множество мелких записей в более крупные батчи, что критически важно для эффективности. Когда MemTable заполняется (обычно при достижении 128-256 МБ), она сбрасывается на диск как файл Level 0.

Файлы Level 0 - это по сути снэпшоты MemTable, записанные в колоночном формате (обычно Parquet или ORC). Они маленькие, неоптимизированные, могут содержать дубликаты и перекрывающиеся диапазоны ключей. Но - и это важно - они доступны для чтения сразу после записи. Это обеспечивает низкую задержку: данные видны в запросах через секунды после поступления.

Параллельно с этим работает процесс компактификации (привет ScyllaDB). Он периодически берет несколько файлов с Level 0 и сливает их в один больший файл на Level 1, попутно удаляя дубликаты и применяя merge-логику (например, суммируя значения для одинаковых ключей). Файлы Level 1, в свою очередь, компактифицируются в Level 2, и так далее. Каждый следующий уровень содержит файлы примерно в 10 раз большего размера.

Эта многоуровневая структура решает известную проблему write amplification ("раздувание записи"). В традиционных B-tree базах данных каждая запись может привести к перезаписи целой страницы или даже перебалансировке дерева. В LSM-tree запись всегда последовательная и происходит только в конец файла, что особенно эффективно для современных SSD и объектных хранилищ типа S3.

Но как же обеспечивается эффективное чтение при такой фрагментированной структуре? Здесь Paimon использует несколько оптимизаций. Во-первых, каждый файл содержит метаданные о диапазоне ключей и значений, которые в нем хранятся. Это позволяет пропускать файлы, которые заведомо не содержат нужных данных. Во-вторых, используются фильтры Блума - вероятностные структуры данных, которые могут быстро сказать "этого ключа точно нет в файле" без необходимости читать сам файл. В-третьих, популярные данные кешируются в памяти, снижая количество обращений к диску.

Unified Changelog: ядро потоковой обработки

Одна из инновационных особенностей Paimon - это концепция unified changelog. В традиционных системах changelog (журнал изменений) и table storage (табличное хранилище) - это разные сущности. У вас есть Kafka для потока событий и отдельно Hive/Iceberg таблица для хранения. Синхронизация между ними - постоянная головная боль.

В Paimon changelog является неотъемлемой частью таблицы. Каждое изменение в таблице автоматически генерирует событие в changelog. При этом changelog не просто дублирует данные - он выводится из самой структуры LSM-tree. Помните процесс компактификации? Когда файлы сливаются, Paimon может определить, какие записи были добавлены, изменены или удалены. Эти изменения и формируют changelog.

Это означает, что вы можете читать одну и ту же таблицу двумя способами. Batch-запросы видят текущий снэпшот - актуальное состояние всех записей. Streaming-запросы подписываются на changelog и получают поток изменений. При этом гарантируется полная консистентность: streaming читатели видят ровно те же изменения, которые видят batch читатели, просто в другой форме. Самая близкая метафора - чтение книги. Batch-режим - это когда вы открываете книгу и читаете текущую версию главы. Streaming-режим - это когда вы следите за правками редактора: "в параграфе 3 изменено слово X на Y", "добавлен новый параграф после параграфа 5". Оба способа дают вам одну и ту же информацию, но оптимизированы для разных сценариев использования.

Ключевые преимущества перед конкурентами: общий обзор

Теперь, когда мы в целом понимаем архитектуру Paimon, давайте сравним его с альтернативными решениями. Это поможет понять, в каких сценариях Paimon действительно лучше всех, а где, возможно, стоит рассмотреть другие варианты.

Начнем с Apache Iceberg и Delta Lake - двух самых популярных open-source lakehouse форматов. Оба изначально создавались для batch-обработки с последующим добавлением streaming возможностей. Это видно в их архитектуре: они используют snapshot-based подход, где каждая транзакция создает новый снэпшот таблицы. Для batch-запросов это отлично работает, но для streaming создает задержки. Типичная задержка в Iceberg/Delta - это минуты, в то время как Paimon обеспечивает секундную задержку.

Кроме того, Iceberg требует отдельного каталога метаданных (например, HMS или Nessie), что усложняет развертывание и управление. Delta Lake привязан к экосистеме Spark/Databricks. В Paimon все метаданные хранятся вместе с данными, и он нативно поддерживается множеством движков: Flink, Spark, Trino, StarRocks.

Apache Hudi позиционируется как streaming-first решение, что делает его ближайшим конкурентом Paimon. Hudi поддерживает два типа таблиц: Copy-on-Write (CoW) оптимизированные для чтения, и Merge-on-Read (MoR) оптимизированные для записи. Проблема в том, что вам нужно заранее выбрать тип таблицы, и изменить его потом сложно. Paimon унифицирует эти подходы: LSM-tree структура автоматически балансирует между скоростью записи и чтения, адаптируясь к нагрузке.

Также Hudi имеет более сложную модель консистентности с различными уровнями изоляции и timeline серверами для координации. Paimon использует более простую, но достаточную для большинства случаев модель: snapshot isolation для batch и at-least-once/exactly-once семантику для streaming.

Сравнение с традиционными streaming storage системами вроде Apache Pulsar или Kafka Streams еще более показательно. Эти системы отлично справляются с потоковой передачей данных, но не предназначены для аналитических запросов. Попробуйте выполнить сложный SQL с несколькими JOIN-ами по Kafka topics - это потребует загрузки всех данных в память или использования внешней системы. Paimon же позволяет выполнять такие запросы напрямую, используя оптимизации колоночного хранения и predicate pushdown.

С другой стороны спектра находятся аналитические СУБД типа ClickHouse или Apache Druid. Они обеспечивают высокую производительность для аналитических запросов, но требуют отдельного ETL-процесса для загрузки данных. Изменение схемы таблицы часто требует полной перезагрузки данных. Paimon поддерживает schema evolution на лету: добавление колонок, изменение типов, даже изменение партиционирования - все это можно делать без остановки потока данных.

Детальные сравнения Apache Paimon с другими форматами

В данной главе я попытаюсь отдельно от предыдущего параграфа детально и более подробно сравнить Apache Paimon с другими форматами данных. Ссылки на все бенчмарки вы можете найти в конце статьи.

1. Apache Paimon против Apache Iceberg

Streaming

По данным компании Alibaba, Paimon демонстрирует задержку менее 100ms для streaming записи, в то время как Iceberg обычно работает с задержками в минуты. Это не недостаток Iceberg - это специфика его архитектуры: Iceberg использует snapshot-based модель, где каждая транзакция создает новый снэпшот.

Batch

Здесь картина меняется: Iceberg показывает 10-20% улучшение производительности по сравнению с Hive для сложных аналитических запросов. Iceberg использует следующие техники оптимизации:

• Hidden partitioning: автоматическое партиционирование по выражениям

• Metadata pruning: агрессивное отсечение файлов по статистике

• Vectorized reads: оптимизированное чтение в колоночных движках

Paimon в batch сценариях требует merge-on-read для primary key таблиц, что может снижать производительность на 15-30% для сложных аналитических запросов.

Модель данных и консистентность

Iceberg:

• ACID транзакции через optimistic concurrency control

• Snapshot isolation для чтения

• Поддержка time travel через snapshot history

• Schema evolution без перезаписи данных

Paimon:

• ACID через двухфазный commit протокол

• Snapshot isolation + continuous changelog

• Time travel + incremental consumption

• Schema evolution в реальном времени для CDC

Ключевое отличие: Paimon поддерживает unified changelog, где каждое изменение доступно как для batch, так и для streaming читателей. Iceberg требует отдельного решения (например, Kafka) для streaming.

Экосистема и совместимость

Iceberg выигрывает в плане поддержки различными движками:

Paimon компенсирует отсутствие поддержки некоторых движков так называемым "Iceberg compatibility mode", позволяя Iceberg-читателям работать с Paimon таблицами.

Когда выбирать Iceberg

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

• Multi-cloud: нужна поддержка Snowflake, BigQuery, Databricks

• Зрелая экосистема со сложной миграцией: требуется максимальная совместимость

• Когда нужна историчность данных: фокус на сложной аналитике с историческими данными, а не на real-time данных

Когда выбирать Paimon

• Стриминг в приоритете: критична низкая задержка

• CDC: real-time синхронизация из операционных БД

• Унифицированное хранение данных: один формат для streaming и batch

• Широкое использование Apache Flink в компании: глубокая интеграция с Apache Flink обеспечивается Apache Paimon'ом.

2. Apache Paimon против Delta Lake

Delta Lake - детище Databricks, тесно интегрированное с их платформой и Apache Spark.

Paimon - открытый проект Apache без привязки к вендору.

Архитектурные различия

Delta Lake:

/delta-table/
  ├── _delta_log/          # Transaction log
  │   ├── 00000.json       # Первая транзакция
  │   ├── 00001.json       # Вторая транзакция
  │   └── 00010.checkpoint.parquet  # Checkp2oint
  └── part-00000-xxx.parquet  # Файлы с данными

Transaction log - это упорядоченная последовательность JSON файлов, описывающих все изменения. Каждые 10 (условно, можно поменять) транзакций создается checkpoint для оптимизации.

Paimon:

/paimon-table/
  ├── manifest/          # Файлы манифеста
  ├── data/              # Организованные в формате LSM-дерева данные
  │   ├── level-0/       # Самые свежие данные
  │   └── level-1/       # Сжатые (компактифицированные) данные
  └── changelog/         # Унифицированный changelog

Структура LSM-дерева принципиально отличается от log-based подхода Delta Lake.

Производительность: бенчмарки

Запись

Независимый бенчмарк 2024 года на датасете 10GB, 100M записей:

Delta Lake быстрее для первоначальной загрузки благодаря простой append-only записи. Paimon эффективнее для инкрементальных обновлений благодаря его LSM-дереву.

Чтение

TPC-DS 1TB benchmark на Spark 3.5 показывает, что при всех "тяжелых" операциях вроде full scan'а, агрегации данных, джойнов и time travel Paimon медленнее на 5-20%, чем Delta Lake, однако в связке с Apache Doris (и materialized views под капотом) использование Paimon дает пятикратный рост производительности по сравнению с использованием Apache Trino + Paimon (бенчмарк от Xiaomi). Так что если у вас уже есть Apache Doris, то Paimon - это для вас.

Когда выбирать Delta Lake

• Если у вас уже есть Databricks

• Тяжелая обработка данных с помощью Spark: Delta Lake оптимизирован именно для Spark

• Самый простой streaming: когда нет сложной логики для real-time данных

• Для любителей Data Governance: Unity Catalog имеет ряд преимуществ перед OpenMetadata (это тема уже для отдельной статьи)

Когда выбирать Paimon

• Нужен опенсорс с полной независимостью от вендора

• Сложный стриминг: если используете CDC и одновременно вам нужна низкая задержка

• Если у вас комбо-опенсорс: Flink + Spark + Trino (или Doris) в одном стеке

3. Apache Paimon против Apache Hudi

Hudi и Paimon - самые близкие конкуренты, оба фокусируются на streaming и incremental processing. Но различия кроятся в деталях.

Архитектура таблиц

Hudi предлагает два типа таблиц:

Copy-on-Write (CoW):

• Данные хранятся в файлах колоночного формата (Parquet)

• Обновление данных создает новые версии файлов.

• CoW оптимален для чтения, а не для записи

Merge-on-Read (MoR):

• Файлы с данными + delta logs

• Обновления данных записываются в row-based delta файлы

• Периодическая компактификация

• MoR оптимален для записи, а не для чтения

Paimon унифицирует подходы:

• LSM-tree автоматически балансирует между чтением и записью

• Нет необходимости выбирать тип таблицы заранее

• Адаптивная компактификация в зависимости от нагрузки

Production бенчмарки

В 2024 году компания Alibaba сделала бенчмарк на следующей тестовой среде: 1 master + 4 core nodes (24 vCPU, 96GB RAM each) Dataset: 500M records, mixed workload (70% updates, 30% inserts). Вот краткие выводы после их тестирования:

Когда выбирать Hudi

• Когда нужна гибкость настройки: множество опций индексации и уровней изоляции

• Много записи, мало чтения: MoR оптимален для write-heavy нагрузок

• Когда нужна интеграция с HBase: у Hudi она есть, у Paimon нет

Когда выбирать Paimon

• Когда нужна автобалансировка нагрузки: если и читаем и пишем одинаково много

• Если используем CDC

• Если есть дефицит ресурсов: Paimon использует меньше памяти и CPU

Сравним в целом по опенсорсу:

Ключевые выводы тестирования, проведенного в Alibaba:

• Потоковая запись и CDC (Streaming Ingestion). За счет использования LSM-дерева (Log-Structured Merge-tree) Paimon превращает случайные обновления (рандомные upsert) в последовательную запись на диск.

• На 100 млн записей Paimon тратит на 30–40% меньше времени на фиксацию батча (commit time), чем Hudi.

• Отсутствует деградация скорости записи при заполнении датасета до финальных 10 GB.

• Память: Paimon требует значительно меньше RAM при стриминге. Hudi на аналогичном потоке может уходить в OOM (Out of Memory) из-за тяжелого процесса индексации в памяти.

• Write Amplification: Благодаря фоновому слиянию мелких файлов (compaction) снижается нагрузка на дисковую подсистему.

• Производительность движков (Spark / Flink):

  • Flink (Streaming): Paimon является нативным решением для Flink, обеспечивая субминутную задержку появления данных для аналитики.

  • Spark (Batch): При расчете агрегатных метрик на слоях DWM/DWS Paimon выигрывает у Hudi за счет встроенных функций слияния (mergeFunction), не требуя написания кастомных сериализаторов.

Сравнительная таблица: все форматы

Базовые характеристики

Производительность

Поддержка движков

Фичи

Где Paimon раскрывает свой потенциал: практические сценарии

Итак, давайте рассмотрим конкретные бизнес-сценарии, где уникальные возможности Paimon дают максимальную отдачу. Это поможет вам понять, подходит ли Paimon для ваших задач.

Real-time DWH для операционной аналитики. Представьте e-commerce платформу, где менеджеры должны видеть метрики продаж с задержкой не более минуты, маркетологи анализируют эффективность кампаний в реальном времени, а финансовый отдел строит отчеты за кварталы и годы. Традиционно это требует сложной архитектуры с отдельными системами для каждого use case. С Paimon все эти сценарии обслуживает одна система. Streaming приложения читают changelog для обновления real-time дашбордов. Batch запросы анализируют исторические снапшоты для глубокой аналитики. Time-travel запросы позволяют восстановить состояние на любой момент времени для аудита или отладки.

CDC и синхронизация разнородных систем. Крупные организации часто имеют зоопарк из различных баз данных: PostgreSQL для веб-приложений, Oracle для ERP, MongoDB для мобильных приложений. Создание единого аналитического слоя поверх этих систем - классическая задача. Paimon естественным образом подходит для этого: CDC-коннекторы захватывают изменения из источников, Paimon таблицы хранят консолидированные данные с полной историей изменений. При этом сохраняется возможность как анализировать текущее состояние, так и отслеживать эволюцию данных во времени.

Feature Store для машинного обучения. ML-модели предъявляют особые требования к данным. Для обучения нужны исторические features с точными временными метками (чтобы избегать утечек данных). Для инференса нужны свежайшие фичи с минимальной задержкой. Paimon таблицы идеально подходят как feature store: исторические фичи читаются через time-travel запросы, real-time (online) фичи - через streaming API, при этом гарантируется консистентность между training и serving.

Event Sourcing и аудит. В регулируемых индустриях (финансы, здравоохранение) часто требуется хранить полную историю всех изменений. Event sourcing паттерн предполагает хранение всех событий, а не только текущего состояния. Paimon с его unified changelog естественным образом поддерживает этот паттерн. Каждое изменение записывается как иммутабельное событие, текущее состояние вычисляется через материализованные представления, а для аудита доступна полная история через time-travel запросы.

IoT и временные ряды. Данные с IoT-устройств имеют специфику: огромный объем, необходимость как real-time мониторинга, так и исторического анализа, частые out-of-order события. Paimon справляется с этими вызовами: партиционирование по времени обеспечивает эффективное хранение, watermark механизм корректно обрабатывает поздние события, а downsampling при компактификации позволяет хранить агрегированные данные для долгосрочного анализа.

Когда Paimon может быть не лучшим выбором

Важно понимать, что не существует универсального решения для всех задач. Есть сценарии, где другие технологии могут быть более подходящими.

Если у вас простая задача загрузки данных раз в день для построения отчетов, и нет требований к real-time, то классические batch форматы типа Parquet файлов или Apache Iceberg могут быть проще в настройке и управлении. Paimon добавляет сложность, которая не оправдана для чисто batch сценариев.

Для сценариев с экстремально высокой нагрузкой по записи (миллионы событий в секунду) и простыми запросами на чтение, специализированные системы типа Apache Kafka или Apache Pulsar могут быть более эффективными. Они оптимизированы именно для такой нагрузки и имеют меньший overhead на запись.

Если основная задача - это сложные аналитические запросы над относительно статичными данными, то колоночные СУБД типа ClickHouse или Apache Druid обеспечат лучшую производительность. Они используют специализированные индексы и форматы хранения, оптимизированные именно для аналитики.

Наконец, если у вас уже есть работающая инфраструктура на базе другого решения, и она удовлетворяет всем требованиям, миграция на Paimon должна быть тщательно обоснована. Любая миграция несет риски и затраты, которые должны окупиться преимуществами новой системы.

Заключение первой части: Paimon как новая парадигма работы с данными (streamhouse)

Apache Paimon представляет собой не просто очередной формат хранения данных, а новый подход к построению data-платформ - так называемый streamhouse. Объединяя streaming и batch парадигмы в единой системе, он устраняет искусственные барьеры между оперативной и исторической аналитикой.

Ключевые инновации Paimon - использование LSM-tree для эффективной записи и чтения, unified changelog для синхронизации streaming и batch представлений, и self-contained архитектура без внешних зависимостей - делают его особенно привлекательным для организаций, строящих современные data-intensive приложения.

Выбор lakehouse формата - это не поиск "лучшего" решения, а поиск наиболее подходящего под ваши требования.

Apache Paimon - отличный выбор для streaming-first архитектур с требованиями к низкой задержке. Если вы строите real-time data platform на базе Flink, Paimon предоставит наилучшую интеграцию и производительность.

Apache Iceberg - зрелое решение для large-scale analytics с широчайшей поддержкой в индустрии. Если вам нужна максимальная совместимость и proven scale - выбирайте Iceberg.

Delta Lake - оптимальное решение для Spark-окружений, особенно если вы уже используете Databricks. Простота и тесная интеграция делают его отличным выбором для быстрого старта.

Apache Hudi - гибкое решение с богатыми возможностями для incremental processing. Если вам нужен fine-grained control над тем, как данные хранятся и обновляются - Hudi предоставит все необходимые инструменты.

Форматы продолжают эволюционировать, и границы между ними размываются. Paimon добавляет совместим с Iceberg, Iceberg постепенно уже улучшает streaming (но все еще не достиг уровня поддержки стриминга как у Paimon), Delta Lake становится более открытым. Возможно, через несколько лет мы увидим конвергенцию к единому стандарту. А пока - выбирайте инструмент под задачу, а не задачу под инструмент.

Часть 2: Практическое руководство с примерами кода

Введение: от теории к практике

В первой части мы разобрали архитектуру Apache Paimon и поняли, как LSM-tree структура позволяет объединить streaming и batch обработку в единой системе. Теперь пришло время применить эти знания на практике. В этой части я покажу вам, как настроить Paimon, организовать потоковую обработку данных и решить типичные задачи, с которыми вы столкнетесь в production.

Важный момент: примеры кода намеренно сбалансированы между простотой для понимания и реалистичностью для production использования. Каждый пример снабжен подробными комментариями, объясняющими не только что делает код, но и почему именно так. Думайте об этих примерах как о строительных блоках, которые вы сможете комбинировать для решения своих задач.

Настройка окружения и создание первой таблицы

Начнем с базовой настройки. Apache Paimon интегрируется с различными вычислительными движками, но наиболее полную поддержку имеет Apache Flink. Это логично, учитывая, что Paimon изначально разрабатывался командой Flink как решение проблемы unified streaming and batch storage. Давайте настроим окружение и создадим первую таблицу.

-- Первый шаг: создание каталога Paimon
-- Каталог - это контейнер для метаданных всех таблиц. Думайте о нем как о базе данных 
-- в традиционной СУБД, но с важным отличием: метаданные хранятся вместе с данными
CREATE CATALOG paimon_catalog WITH (
    'type' = 'paimon',
    -- warehouse - корневая директория для всех данных каталога
    -- Может быть локальной файловой системой для разработки или S3/HDFS для production
    'warehouse' = 's3://your-bucket/paimon-warehouse',
    -- Опционально: настройки для S3 (если используете AWS)
    's3.endpoint' = 'https://s3.amazonaws.com',
    's3.access-key' = '${env:AWS_ACCESS_KEY_ID}',
    's3.secret-key' = '${env:AWS_SECRET_ACCESS_KEY}'
);

-- Переключаемся на созданный каталог
USE CATALOG paimon_catalog;

-- Создаем базу данных (namespace для таблиц)
CREATE DATABASE IF NOT EXISTS production;
USE production;

-- Теперь создадим таблицу с продуманной структурой
-- Это таблица событий пользователей - типичный сценарий для streaming
CREATE TABLE user_events (
    -- Основные поля данных
    event_id BIGINT,
    user_id BIGINT,
    event_type STRING,
    event_time TIMESTAMP(3),  -- (3) означает миллисекундную точность
    
    -- JSON поле для гибкого хранения свойств события
    -- Это позволяет добавлять новые атрибуты без изменения схемы
    properties STRING,
    
    -- Вложенная структура для информации об устройстве
    device ROW<
        type STRING,
        os STRING,
        browser STRING,
        ip STRING
    >,
    
    -- Системные поля для отслеживания
    processing_time TIMESTAMP(3) METADATA FROM 'timestamp',  -- когда запись была добавлена
    
    -- Primary key определяет уникальность и порядок сортировки внутри файлов
    -- Это критически важно для производительности и семантики обновлений
    PRIMARY KEY (user_id, event_id) NOT ENFORCED,
    
    -- Watermark для обработки поздних событий в streaming режиме
    -- Разрешаем задержку до 10 секунд
    WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) 
-- Партиционирование по дню события для эффективного pruning при запросах
PARTITIONED BY (event_day STRING) 
WITH (
    -- Количество bucket'ов определяет параллелизм записи
    -- Выбирайте по следующему принципу: (ожидаемый объем данных в секунду) / (128 MB)
    'bucket' = '8',
    
    -- Ключевой параметр: как генерировать changelog
    -- full-compaction: changelog создается при компактификации (лучше для больших объемов)
    -- input: changelog создается сразу при записи (лучше для низкой latency)
    'changelog-producer' = 'full-compaction',
    
    -- Частота компактификации - баланс между свежестью changelog и нагрузкой
    'changelog-producer.compaction-interval' = '2 min',
    
    -- Размер буфера в памяти перед сбросом на диск
    -- Больший буфер = меньше файлов, но выше latency
    'write-buffer-size' = '256 MB',
    
    -- Стратегия слияния при обновлении существующих записей
    'merge-engine' = 'deduplicate',  -- оставляем только последнюю версию
    
    -- Настройки компактификации для оптимизации размера файлов
    'compaction.min-file-num' = '3',  -- начинаем компактификацию при 3+ файлах
    'compaction.max-file-num' = '10',  -- но не ждем больше 10 файлов
    
    -- Сколько снапшотов хранить для time-travel запросов
    'snapshot.expire.limit' = '100',
    'snapshot.expire.execution-mode' = 'async'  -- чистка в фоне
);

-- Создаем вычисляемое поле через виртуальную колонку
-- Это эффективнее, чем вычислять при каждом запросе
ALTER TABLE user_events ADD event_hour AS HOUR(event_time);

Обратите внимание на несколько ключевых моментов в этой конфигурации. Во-первых, выбор первичного ключа критически важен - он определяет не только уникальность записей, но и то, как данные будут физически организованы в файлах. Записи с близкими значениями первичных ключей будут храниться рядом, что ускоряет range-запросы. Во-вторых, настройка bucket'ов влияет на параллелизм - каждый bucket может записываться независимо, но слишком много bucket'ов приведет к фрагментации данных. В-третьих, выбор между 'input' и 'full-compaction' для changelog-producer - это компромисс между latency и эффективностью.

Потоковая запись данных из Kafka

Теперь давайте реализуем типичный production сценарий: чтение событий из Kafka, их обогащение и запись в Paimon таблицу. Это покажет, как Paimon интегрируется с существующей streaming инфраструктурой.

// Java код для Flink DataStream API
// Этот подход дает больше контроля, чем чистый SQL

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.types.Row;
import java.time.Duration;

public class KafkaToPaimonPipeline {
    
    public static void main(String[] args) throws Exception {
        // Настройка окружения Flink с оптимальными параметрами для production
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Checkpoint'ы критически важны для exactly-once семантики
        // Без них при сбое можем потерять или дублировать данные
        env.enableCheckpointing(60000);  // каждые 60 секунд
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);  // минимум 30 сек между checkpoint'ами
        env.getCheckpointConfig().setCheckpointTimeout(120000);  // таймаут 2 минуты
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);  // только один checkpoint одновременно
        
        // Table API окружение для работы с SQL
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        
        // Настройка idle timeout для обработки редких событий
        // Без этого watermark может "застрять" на партициях без данных
        tableEnv.getConfig().set(
            "table.exec.source.idle-timeout", 
            Duration.ofMinutes(1).toString()
        );
        
        // Создаем источник данных - Kafka topic
        tableEnv.executeSql("""
            CREATE TEMPORARY TABLE kafka_events (
                -- Поля из Kafka сообщения
                event_id BIGINT,
                user_id BIGINT,
                event_type STRING,
                event_timestamp BIGINT,  -- Unix timestamp в миллисекундах
                properties STRING,  -- JSON строка
                
                -- Метаданные Kafka для отладки и мониторинга
                kafka_timestamp TIMESTAMP(3) METADATA FROM 'timestamp',
                kafka_partition INT METADATA FROM 'partition',
                kafka_offset BIGINT METADATA FROM 'offset',
                
                -- Преобразуем Unix timestamp в TIMESTAMP для watermark
                event_time AS TO_TIMESTAMP_LTZ(event_timestamp, 3),
                WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
            ) WITH (
                'connector' = 'kafka',
                'topic' = 'user-events-raw',
                'properties.bootstrap.servers' = 'kafka-broker-1:9092,kafka-broker-2:9092',
                'properties.group.id' = 'paimon-ingestion-${env:ENVIRONMENT}',
                
                -- Стратегия чтения: начинаем с последнего committed offset
                'scan.startup.mode' = 'group-offsets',
                -- Если offset'ов нет (первый запуск), начинаем с начала
                'properties.auto.offset.reset' = 'earliest',
                
                'format' = 'json',
                'json.fail-on-missing-field' = 'false',  -- не падаем на отсутствующих полях
                'json.ignore-parse-errors' = 'false'  -- но падаем на невалидном JSON
            )
        """);
        
        // Справочная таблица для обогащения (также Paimon)
        // Демонстрирует паттерн lookup join
        tableEnv.executeSql("""
            CREATE TABLE user_profiles (
                user_id BIGINT,
                username STRING,
                email STRING,
                registration_date DATE,
                user_segment STRING,
                country STRING,
                -- Техническое поле для версионирования справочника
                updated_at TIMESTAMP(3),
                PRIMARY KEY (user_id) NOT ENFORCED
            ) WITH (
                'connector' = 'paimon',
                -- Важно: включаем continuous discovery для свежих данных
                'continuous.discovery-interval' = '30 s',
                -- Кэшируем lookup результаты для производительности
                'lookup.cache.ttl' = '10 min'
            )
        """);
        
        // Таблица для подозрительных событий (демонстрирует side output)
        tableEnv.executeSql("""
            CREATE TABLE suspicious_events (
                event_id BIGINT,
                user_id BIGINT,
                event_type STRING,
                reason STRING,
                detection_time TIMESTAMP(3),
                PRIMARY KEY (event_id) NOT ENFORCED
            ) WITH (
                'connector' = 'paimon',
                'merge-engine' = 'deduplicate'
            )
        """);
        
        // Основной ETL pipeline с обогащением и валидацией
        String mainPipeline = """
            -- Используем STATEMENT SET для атомарного выполнения нескольких INSERT
            EXECUTE STATEMENT SET
            BEGIN
                -- Основной поток: обогащенные валидные события
                INSERT INTO user_events
                SELECT 
                    ke.event_id,
                    ke.user_id,
                    ke.event_type,
                    ke.event_time,
                    
                    -- Обогащаем properties информацией из профиля
                    -- JSON_MERGE не существует в Flink SQL, используем конкатенацию
                    CONCAT(
                        SUBSTR(ke.properties, 1, LENGTH(ke.properties) - 1),
                        ',',
                        '"username":"', COALESCE(up.username, ''), '",',
                        '"user_segment":"', COALESCE(up.user_segment, ''), '",',
                        '"country":"', COALESCE(up.country, ''), '",',
                        '"account_age_days":', 
                        CAST(TIMESTAMPDIFF(DAY, up.registration_date, CURRENT_DATE) AS STRING),
                        '}'
                    ) as properties,
                    
                    -- Парсим device информацию из JSON
                    ROW(
                        JSON_VALUE(ke.properties, '$.device.type'),
                        JSON_VALUE(ke.properties, '$.device.os'),
                        JSON_VALUE(ke.properties, '$.device.browser'),
                        JSON_VALUE(ke.properties, '$.device.ip')
                    ) as device,
                    
                    -- Вычисляемое поле для партиционирования
                    DATE_FORMAT(ke.event_time, 'yyyy-MM-dd') as event_day
                    
                FROM kafka_events ke
                -- Temporal join для получения актуального профиля на момент события
                LEFT JOIN user_profiles FOR SYSTEM_TIME AS OF ke.event_time AS up
                    ON ke.user_id = up.user_id
                -- Фильтруем только валидные события
                WHERE ke.event_type IS NOT NULL
                    AND ke.user_id IS NOT NULL
                    AND ke.event_id IS NOT NULL;
                
                -- Side output: подозрительные события для отдельного анализа
                INSERT INTO suspicious_events
                SELECT 
                    event_id,
                    user_id,
                    event_type,
                    CASE 
                        -- Определяем причину подозрительности
                        WHEN event_type = 'purchase' AND 
                             CAST(JSON_VALUE(properties, '$.amount') AS DECIMAL(10,2)) > 10000 
                             THEN 'high_value_transaction'
                        WHEN event_type IN ('login', 'password_change') AND 
                             HOUR(event_time) BETWEEN 2 AND 5 
                             THEN 'unusual_time_activity'
                        WHEN JSON_VALUE(properties, '$.failed_attempts') > '5' 
                             THEN 'multiple_failed_attempts'
                        ELSE 'other'
                    END as reason,
                    CURRENT_TIMESTAMP as detection_time
                FROM kafka_events
                WHERE 
                    -- Условия для подозрительных событий
                    (event_type = 'purchase' AND 
                     CAST(JSON_VALUE(properties, '$.amount') AS DECIMAL(10,2)) > 10000)
                    OR (event_type IN ('login', 'password_change') AND 
                        HOUR(event_time) BETWEEN 2 AND 5)
                    OR CAST(JSON_VALUE(properties, '$.failed_attempts') AS INT) > 5;
            END
        """;
        
        // Выполняем pipeline
        tableEnv.executeSql(mainPipeline);
        
        // Запускаем job
        env.execute("Kafka to Paimon Streaming ETL");
    }
}

Этот пример демонстрирует несколько важных паттернов. Temporal join позволяет обогащать поток данными из справочника с учетом времени события - мы получаем версию профиля, актуальную на момент события, а не текущую. Side output через STATEMENT SET позволяет разделить поток на несколько целевых таблиц в одной транзакции. Обработка JSON через встроенные функции показывает, как работать с semi-structured данными без необходимости заранее определять жесткую схему.

Гибридные запросы: одни данные, разные режимы чтения

Одна из уникальных особенностей Paimon - возможность читать одни и те же данные как в streaming, так и в batch режиме. Давайте посмотрим, как это работает на практике и какие возможности открывает.

# Python код с использованием PyFlink
# Демонстрирует различные режимы работы с одной таблицей

from pyflink.table import EnvironmentSettings, TableEnvironment
from pyflink.table.expressions import col, lit, call
from datetime import datetime, timedelta
import json

class HybridAnalytics:
    """
    Класс демонстрирует различные способы чтения Paimon таблиц
    для разных аналитических сценариев
    """
    
    def __init__(self, warehouse_path: str):
        """
        Инициализация окружений для batch и streaming обработки
        
        Важный момент: мы создаем два отдельных окружения, потому что
        batch и streaming режимы имеют разную семантику выполнения
        """
        # Batch окружение для аналитических запросов
        self.batch_env = TableEnvironment.create(
            EnvironmentSettings.in_batch_mode()
        )
        
        # Streaming окружение для real-time обработки
        self.stream_env = TableEnvironment.create(
            EnvironmentSettings.in_streaming_mode()
        )
        
        # Настраиваем каталог для обоих окружений
        for env in [self.batch_env, self.stream_env]:
            env.execute_sql(f"""
                CREATE CATALOG paimon_catalog WITH (
                    'type' = 'paimon',
                    'warehouse' = '{warehouse_path}'
                )
            """)
            env.use_catalog("paimon_catalog")
            env.use_database("production")
    
    def batch_analytics(self) -> None:
        """
        Batch аналитика: сложные агрегации над историческими данными
        
        В batch режиме Paimon читает последний снапшот таблицы,
        оптимизируя запрос для минимального количества I/O операций
        """
        
        # Сложный аналитический запрос с window функциями
        # Такие запросы эффективны в batch режиме благодаря columnar storage
        result = self.batch_env.sql_query("""
            WITH user_metrics AS (
                -- Первый CTE: базовые метрики по пользователям
                SELECT 
                    user_id,
                    DATE(event_time) as event_date,
                    COUNT(*) as daily_events,
                    COUNT(DISTINCT event_type) as unique_event_types,
                    
                    -- Используем MAP_AGG для группировки событий по типам
                    -- Это эффективнее, чем множественные подзапросы
                    MAP_AGG(event_type, event_id) as events_by_type,
                    
                    -- Window функции для трендов
                    COUNT(*) OVER (
                        PARTITION BY user_id 
                        ORDER BY DATE(event_time) 
                        ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
                    ) as weekly_events,
                    
                    -- Находим самое популярное событие за день
                    FIRST_VALUE(event_type) OVER (
                        PARTITION BY user_id, DATE(event_time)
                        ORDER BY COUNT(*) DESC
                    ) as most_frequent_event
                    
                FROM user_events
                WHERE event_time >= CURRENT_TIMESTAMP - INTERVAL '30' DAY
                GROUP BY user_id, DATE(event_time), event_type
            ),
            cohort_analysis AS (
                -- Второй CTE: когортный анализ
                SELECT 
                    up.registration_date,
                    um.event_date,
                    -- Дни с момента регистрации
                    DATEDIFF(um.event_date, up.registration_date) as days_since_registration,
                    COUNT(DISTINCT um.user_id) as active_users,
                    AVG(um.daily_events) as avg_events_per_user
                FROM user_metrics um
                JOIN user_profiles up ON um.user_id = up.user_id
                GROUP BY up.registration_date, um.event_date
            )
            -- Финальный запрос: retention матрица
            SELECT 
                registration_date as cohort,
                days_since_registration,
                active_users,
                -- Рассчитываем retention rate
                active_users * 100.0 / FIRST_VALUE(active_users) OVER (
                    PARTITION BY registration_date 
                    ORDER BY days_since_registration
                ) as retention_rate,
                avg_events_per_user
            FROM cohort_analysis
            WHERE days_since_registration BETWEEN 0 AND 30
            ORDER BY registration_date, days_since_registration
        """)
        
        # Batch запросы возвращают полный результат, который можно сохранить
        result.execute_insert("cohort_retention_analysis")
        print("Cohort analysis completed and saved")
    
    def streaming_monitoring(self) -> None:
        """
        Streaming мониторинг: real-time метрики и алерты
        
        В streaming режиме Paimon читает changelog и continuously
        обновляет результаты по мере поступления новых данных
        """
        
        # Создаем continuous view для real-time метрик
        # TUMBLE window создает non-overlapping временные окна
        self.stream_env.execute_sql("""
            CREATE TEMPORARY VIEW realtime_metrics AS
            SELECT 
                -- Временное окно
                TUMBLE_START(event_time, INTERVAL '1' MINUTE) as window_start,
                TUMBLE_END(event_time, INTERVAL '1' MINUTE) as window_end,
                
                -- Метрики в окне
                COUNT(*) as total_events,
                COUNT(DISTINCT user_id) as unique_users,
                COUNT(DISTINCT event_type) as event_types,
                
                -- Распределение по типам событий
                MAP_AGG(event_type, COUNT(*)) as event_distribution,
                
                -- P95 latency (время между событием и обработкой)
                PERCENTILE_CONT(
                    EXTRACT(EPOCH FROM processing_time - event_time), 
                    0.95
                ) as p95_latency_seconds,
                
                -- Детекция аномалий: всплеск активности
                CASE 
                    WHEN COUNT(*) > (
                        -- Сравниваем с средним за последний час
                        SELECT AVG(cnt) * 2
                        FROM (
                            SELECT COUNT(*) as cnt
                            FROM user_events
                            WHERE event_time >= CURRENT_TIMESTAMP - INTERVAL '1' HOUR
                            GROUP BY TUMBLE(event_time, INTERVAL '1' MINUTE)
                        )
                    ) THEN TRUE
                    ELSE FALSE
                END as is_anomaly
                
            FROM user_events
            GROUP BY TUMBLE(event_time, INTERVAL '1' MINUTE)
        """)
        
        # Материализуем метрики в Paimon таблицу для исторического анализа
        self.stream_env.execute_sql("""
            CREATE TABLE IF NOT EXISTS monitoring_metrics (
                window_start TIMESTAMP(3),
                window_end TIMESTAMP(3),
                total_events BIGINT,
                unique_users BIGINT,
                event_types INT,
                event_distribution STRING,  -- Изменено с MAP на STRING
                p95_latency_seconds DOUBLE,
                is_anomaly BOOLEAN,
                -- Добавляем время записи для аудита
                ingestion_time TIMESTAMP(3) METADATA FROM 'timestamp',
                PRIMARY KEY (window_start) NOT ENFORCED
            ) WITH (
                'merge-engine' = 'deduplicate',
                -- Храним только последнюю версию метрик для каждого окна
                'changelog-producer' = 'none',
                -- Агрессивная компактификация для таблицы метрик
                'compaction.max-file-num' = '5'
            )
        """)
        
        # Записываем метрики
        self.stream_env.execute_sql("""
            INSERT INTO monitoring_metrics
            SELECT * FROM realtime_metrics
        """)
        
        # Создаем алерты на основе метрик
        alerts = self.stream_env.sql_query("""
            SELECT 
                window_start,
                'ANOMALY_DETECTED' as alert_type,
                CONCAT(
                    'Unusual activity detected: ',
                    CAST(total_events AS STRING),
                    ' events in window'
                ) as message,
                MAP[
                    'total_events', CAST(total_events AS STRING),
                    'unique_users', CAST(unique_users AS STRING),
                    'event_distribution', CAST(event_distribution AS STRING)
                ] as context
            FROM realtime_metrics
            WHERE is_anomaly = TRUE
        """)
        
        # В production здесь была бы отправка в Kafka/Slack/PagerDuty
        alerts.execute_insert("alerts_table")
    
    def time_travel_analysis(self, hours_ago: int = 1) -> None:
        """
        Time Travel запросы: анализ исторических снапшотов
        
        Paimon сохраняет историю снапшотов, позволяя запрашивать
        состояние таблицы на любой момент времени в прошлом
        """
        
        # Вычисляем timestamp для time travel
        target_timestamp = int(
            (datetime.now() - timedelta(hours=hours_ago)).timestamp() * 1000
        )
        
        # Time travel через SQL hint
        historical_data = self.batch_env.sql_query(f"""
            -- OPTIONS hint позволяет передать параметры сканирования
            SELECT 
                user_id,
                COUNT(*) as event_count,
                COLLECT(event_type) as event_sequence,
                MIN(event_time) as first_event,
                MAX(event_time) as last_event
            FROM user_events 
            /*+ OPTIONS(
                'scan.timestamp-millis' = '{target_timestamp}',
                -- Можно также использовать snapshot-id вместо timestamp
                -- 'scan.snapshot-id' = '12345'
            ) */
            WHERE user_id IN (
                -- Находим активных пользователей в тот момент времени
                SELECT DISTINCT user_id 
                FROM user_events 
                /*+ OPTIONS('scan.timestamp-millis' = '{target_timestamp}') */
                WHERE event_time >= TIMESTAMP '{datetime.now() - timedelta(hours=hours_ago+1)}'
                  AND event_time <= TIMESTAMP '{datetime.now() - timedelta(hours=hours_ago)}'
                LIMIT 100
            )
            GROUP BY user_id
        """)
        
        # Сравнение с текущим состоянием для анализа изменений
        comparison = self.batch_env.sql_query(f"""
            WITH historical AS (
                SELECT user_id, COUNT(*) as historical_count
                FROM user_events 
                /*+ OPTIONS('scan.timestamp-millis' = '{target_timestamp}') */
                GROUP BY user_id
            ),
            current AS (
                SELECT user_id, COUNT(*) as current_count
                FROM user_events
                GROUP BY user_id
            )
            SELECT 
                COALESCE(h.user_id, c.user_id) as user_id,
                COALESCE(h.historical_count, 0) as events_then,
                COALESCE(c.current_count, 0) as events_now,
                COALESCE(c.current_count, 0) - COALESCE(h.historical_count, 0) as events_added,
                CASE 
                    WHEN h.historical_count IS NULL THEN 'new_user'
                    WHEN c.current_count IS NULL THEN 'churned_user'
                    WHEN c.current_count > h.historical_count * 1.5 THEN 'growing_activity'
                    WHEN c.current_count < h.historical_count * 0.5 THEN 'declining_activity'
                    ELSE 'stable_activity'
                END as user_status
            FROM historical h
            FULL OUTER JOIN current c ON h.user_id = c.user_id
            WHERE COALESCE(c.current_count, 0) - COALESCE(h.historical_count, 0) != 0
            ORDER BY ABS(COALESCE(c.current_count, 0) - COALESCE(h.historical_count, 0)) DESC
            LIMIT 100
        """)
        
        print(f"Time travel analysis for {hours_ago} hours ago completed")
        return comparison
    
    def incremental_processing(self) -> None:
        """
        Инкрементальная обработка: читаем только новые данные с последнего запуска
        
        Это ключевой паттерн для эффективной batch обработки больших таблиц
        """
        
        # Получаем последний обработанный snapshot из таблицы состояния
        last_processed = self.batch_env.sql_query("""
            SELECT MAX(last_snapshot_id) as snapshot_id
            FROM processing_state
            WHERE job_name = 'daily_aggregation'
        """).collect()[0][0] or 0
        
        # Читаем только изменения с последнего snapshot
        incremental_data = self.batch_env.sql_query(f"""
            SELECT 
                user_id,
                DATE(event_time) as event_date,
                event_type,
                COUNT(*) as event_count,
                -- Получаем текущий snapshot для сохранения состояния
                _SNAPSHOT_ID_ as snapshot_id
            FROM user_events
            /*+ OPTIONS(
                'scan.mode' = 'incremental',
                'incremental-between' = '{last_processed},latest'
            ) */
            GROUP BY user_id, DATE(event_time), event_type, _SNAPSHOT_ID_
        """)
        
        # Обрабатываем инкрементальные данные
        self.batch_env.execute_sql("""
            -- MERGE INTO для upsert логики
            MERGE INTO daily_user_aggregates target
            USING (
                SELECT 
                    user_id,
                    event_date,
                    SUM(event_count) as total_events,
                    MAP_AGG(event_type, event_count) as events_by_type,
                    MAX(snapshot_id) as snapshot_id
                FROM incremental_data
                GROUP BY user_id, event_date
            ) source
            ON target.user_id = source.user_id 
               AND target.event_date = source.event_date
            WHEN MATCHED THEN
                UPDATE SET 
                    total_events = target.total_events + source.total_events,
                    -- Merge maps: суммируем значения для одинаковых ключей
                    events_by_type = MAP_UNION_SUM(
                        target.events_by_type, 
                        source.events_by_type
                    ),
                    last_updated = CURRENT_TIMESTAMP
            WHEN NOT MATCHED THEN
                INSERT (user_id, event_date, total_events, events_by_type, last_updated)
                VALUES (
                    source.user_id, 
                    source.event_date, 
                    source.total_events, 
                    source.events_by_type,
                    CURRENT_TIMESTAMP
                )
        """)
        
        # Обновляем состояние обработки
        self.batch_env.execute_sql(f"""
            INSERT INTO processing_state (job_name, last_snapshot_id, processed_at)
            VALUES (
                'daily_aggregation',
                (SELECT MAX(snapshot_id) FROM incremental_data),
                CURRENT_TIMESTAMP
            )
        """)
        
        print("Incremental processing completed")

# Использование класса
if __name__ == "__main__":
    analytics = HybridAnalytics("s3://your-bucket/paimon-warehouse")
    
    # Запускаем различные типы анализа
    analytics.batch_analytics()  # Сложная аналитика над историческими данными
    analytics.streaming_monitoring()  # Real-time мониторинг
    analytics.time_travel_analysis(hours_ago=24)  # Анализ вчерашнего состояния
    analytics.incremental_processing()  # Эффективная обработка новых данных

Этот пример показывает мощь гибридной архитектуры Paimon. Обратите внимание, как одна и та же таблица user_events используется для совершенно разных сценариев: batch аналитика читает полный снапшот для сложных агрегаций, streaming мониторинг следит за изменениями в реальном времени, time travel позволяет анализировать историческое состояние, а инкрементальная обработка эффективно работает только с новыми данными.

CDC сценарий: создание аналитической реплики с историей

Change Data Capture (CDC) - это паттерн захвата изменений из операционных баз данных для аналитики. Paimon естественным образом поддерживает CDC благодаря своей LSM-tree архитектуре и unified changelog. Давайте реализуем полноценный CDC pipeline с сохранением истории изменений:

-- SQL скрипт для настройки CDC pipeline от MySQL к Paimon
-- с полным версионированием и аудитом изменений

-- Шаг 1: Создаем источник CDC из MySQL
-- Используем Flink CDC connector для чтения binlog
CREATE TEMPORARY TABLE mysql_orders_cdc (
    -- Поля из исходной таблицы
    order_id BIGINT,
    user_id BIGINT,
    product_id BIGINT,
    quantity INT,
    price DECIMAL(10, 2),
    order_status STRING,
    payment_method STRING,
    shipping_address STRING,
    created_at TIMESTAMP(3),
    updated_at TIMESTAMP(3),
    
    -- Метаданные CDC для отслеживания типа операции
    -- Это критически важно для правильной обработки удалений
    op_type STRING METADATA FROM 'op_type' VIRTUAL,
    op_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    
    -- Primary key должен соответствовать исходной таблице
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '${env:MYSQL_HOST}',
    'port' = '3306',
    'username' = '${env:MYSQL_USER}',
    'password' = '${env:MYSQL_PASSWORD}',
    'database-name' = 'ecommerce',
    'table-name' = 'orders',
    
    -- Важно: начинаем с полного снапшота, затем переключаемся на binlog
    -- Это гарантирует, что мы не пропустим существующие данные
    'scan.startup.mode' = 'initial',
    
    -- Настройки для production
    'server-id' = '5400-5404',  -- диапазон для параллельного чтения
    'scan.snapshot.fetch.size' = '1024',  -- размер батча при снапшоте
    'connect.timeout' = '30s',
    'server-time-zone' = 'UTC',
    
    -- Debezium properties для тонкой настройки
    'debezium.snapshot.locking.mode' = 'none',  -- не блокируем таблицы
    'debezium.include.schema.changes' = 'false'  -- не отслеживаем DDL
);

-- Шаг 2: Целевая Paimon таблица с полной историей версий
-- Эта таблица хранит все версии каждой записи
CREATE TABLE IF NOT EXISTS orders_history (
    -- Исходные поля
    order_id BIGINT,
    user_id BIGINT,
    product_id BIGINT,
    quantity INT,
    price DECIMAL(10, 2),
    order_status STRING,
    payment_method STRING,
    shipping_address STRING,
    
    -- Временные метки из источника
    created_at TIMESTAMP(3),
    updated_at TIMESTAMP(3),
    
    -- Поля версионирования
    version_number BIGINT,  -- инкрементальный номер версии
    valid_from TIMESTAMP(3),  -- когда версия стала актуальной
    valid_to TIMESTAMP(3),  -- когда версия перестала быть актуальной (NULL для текущей)
    is_current BOOLEAN,  -- флаг текущей версии
    
    -- Метаданные операции
    operation_type STRING,  -- INSERT, UPDATE, DELETE, SNAPSHOT
    operation_timestamp TIMESTAMP(3),  -- когда произошла операция
    operation_user STRING,  -- кто выполнил операцию (если доступно)
    
    -- Аудит
    ingestion_timestamp TIMESTAMP(3) METADATA FROM 'timestamp',  -- когда записано в Paimon
    
    -- Составной primary key для хранения всех версий
    PRIMARY KEY (order_id, version_number) NOT ENFORCED
) 
PARTITIONED BY (DATE_FORMAT(valid_from, 'yyyy-MM')) 
WITH (
    -- Используем partial-update для эффективного обновления полей
    'merge-engine' = 'partial-update',
    'partial-update.ignore-delete' = 'false',
    
    -- Changelog настройки для streaming читателей
    'changelog-producer' = 'input',
    'changelog-producer.row-deduplicate' = 'false',  -- сохраняем все версии
    
    -- Оптимизация для версионированных данных
    'bucket' = '16',  -- больше bucket'ов для параллельной записи
    'bucket-key' = 'order_id',  -- распределение по order_id
    
    -- Компактификация
    'compaction.min-file-num' = '5',
    'compaction.max-file-num' = '10',
    'compaction.optimization-interval' = '1 h',  -- оптимизация каждый час
    
    -- Хранение снапшотов для time travel
    'snapshot.time-retained' = '7 d',  -- храним снапшоты 7 дней
    'snapshot.num-retained.min' = '10',
    'snapshot.num-retained.max' = '100'
);

-- Шаг 3: Текущее состояние (SCD Type 1) для быстрых запросов
-- Отдельная таблица только с актуальными версиями
CREATE TABLE IF NOT EXISTS orders_current (
    order_id BIGINT,
    user_id BIGINT,
    product_id BIGINT,
    quantity INT,
    price DECIMAL(10, 2),
    order_status STRING,
    payment_method STRING,
    shipping_address STRING,
    created_at TIMESTAMP(3),
    updated_at TIMESTAMP(3),
    
    -- Метаданные последнего изменения
    last_operation STRING,
    last_modified TIMESTAMP(3),
    
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'merge-engine' = 'deduplicate',  -- храним только последнюю версию
    'changelog-producer' = 'full-compaction',
    'changelog-producer.compaction-interval' = '5 min'
);

-- Шаг 4: ETL логика с обработкой всех типов CDC событий
-- Используем Table API для сложной логики версионирования
CREATE TEMPORARY VIEW orders_with_versions AS
WITH version_calc AS (
    -- Вычисляем номер версии для каждой записи
    SELECT 
        *,
        -- ROW_NUMBER дает нам инкрементальный номер версии
        ROW_NUMBER() OVER (
            PARTITION BY order_id 
            ORDER BY op_ts
        ) as version_number
    FROM mysql_orders_cdc
),
versioned_data AS (
    -- Добавляем временные границы для каждой версии
    SELECT 
        order_id,
        user_id,
        product_id,
        quantity,
        price,
        order_status,
        payment_method,
        shipping_address,
        created_at,
        updated_at,
        version_number,
        op_ts as valid_from,
        -- valid_to - это valid_from следующей версии
        LEAD(op_ts) OVER (
            PARTITION BY order_id 
            ORDER BY version_number
        ) as valid_to,
        -- Текущая версия имеет valid_to = NULL
        CASE 
            WHEN LEAD(op_ts) OVER (PARTITION BY order_id ORDER BY version_number) IS NULL 
            THEN TRUE 
            ELSE FALSE 
        END as is_current,
        -- Определяем тип операции
        CASE op_type
            WHEN '+I' THEN 'INSERT'
            WHEN '-U' THEN 'UPDATE_BEFORE'
            WHEN '+U' THEN 'UPDATE_AFTER'
            WHEN '-D' THEN 'DELETE'
            ELSE 'UNKNOWN'
        END as operation_type,
        op_ts as operation_timestamp,
        -- Можно добавить информацию о пользователе из контекста
        COALESCE(SYSTEM_USER(), 'system') as operation_user
    FROM version_calc
)
SELECT * FROM versioned_data
-- Фильтруем UPDATE_BEFORE события, оставляем только UPDATE_AFTER
WHERE operation_type != 'UPDATE_BEFORE';

-- Вставка в историческую таблицу
INSERT INTO orders_history
SELECT 
    order_id,
    user_id,
    product_id,
    quantity,
    price,
    order_status,
    payment_method,
    shipping_address,
    created_at,
    updated_at,
    version_number,
    valid_from,
    valid_to,
    is_current,
    operation_type,
    operation_timestamp,
    operation_user
FROM orders_with_versions;

-- Вставка/обновление текущего состояния
INSERT INTO orders_current
SELECT 
    order_id,
    user_id,
    product_id,
    quantity,
    price,
    order_status,
    payment_method,
    shipping_address,
    created_at,
    updated_at,
    operation_type as last_operation,
    operation_timestamp as last_modified
FROM orders_with_versions
WHERE is_current = TRUE;

-- Шаг 5: Создаем материализованные представления для аналитики
CREATE TABLE IF NOT EXISTS order_status_transitions (
    order_id BIGINT,
    from_status STRING,
    to_status STRING,
    transition_time TIMESTAMP(3),
    time_in_status BIGINT,  -- секунды в предыдущем статусе
    PRIMARY KEY (order_id, transition_time) NOT ENFORCED
) WITH ('changelog-producer' = 'none');

-- Заполняем таблицу переходов статусов
INSERT INTO order_status_transitions
SELECT 
    order_id,
    LAG(order_status) OVER (PARTITION BY order_id ORDER BY version_number) as from_status,
    order_status as to_status,
    valid_from as transition_time,
    TIMESTAMPDIFF(
        SECOND, 
        LAG(valid_from) OVER (PARTITION BY order_id ORDER BY version_number),
        valid_from
    ) as time_in_status
FROM orders_history
WHERE operation_type IN ('INSERT', 'UPDATE_AFTER')
    AND LAG(order_status) OVER (PARTITION BY order_id ORDER BY version_number) != order_status;

-- Шаг 6: Запросы для анализа версионированных данных
-- Пример: История изменений конкретного заказа
SELECT 
    version_number,
    operation_type,
    order_status,
    price,
    quantity,
    valid_from,
    valid_to,
    CASE 
        WHEN valid_to IS NULL THEN 'Current'
        ELSE CONCAT(
            'Historical (', 
            CAST(TIMESTAMPDIFF(HOUR, valid_to, CURRENT_TIMESTAMP) AS STRING),
            ' hours ago)'
        )
    END as version_status
FROM orders_history
WHERE order_id = 12345
ORDER BY version_number DESC;

-- Пример: Аудит всех удаленных заказов за последние 24 часа
SELECT 
    order_id,
    user_id,
    order_status,
    price,
    operation_timestamp as deleted_at,
    operation_user as deleted_by
FROM orders_history
WHERE operation_type = 'DELETE'
    AND operation_timestamp >= CURRENT_TIMESTAMP - INTERVAL '24' HOUR
ORDER BY operation_timestamp DESC;

Этот CDC пример демонстрирует полноценное решение для версионирования данных. Мы сохраняем полную историю изменений, отслеживаем переходы между состояниями и можем восстановить данные на любой момент времени. Важно понимать, что такой подход требует больше места для хранения, но дает полный аудит и возможность анализа эволюции данных.

Оптимизация производительности и мониторинг

Завершим практическую часть рассмотрением оптимизации и мониторинга Paimon таблиц. Производительность зависит от множества факторов: размера файлов, частоты компактификации, количества уровней в LSM-tree. Давайте разберем, как настраивать эти параметры и отслеживать состояние системы.

// Java код для мониторинга и оптимизации Paimon таблиц

import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.operation.metrics.MetricRegistry;
import java.util.*;
import java.util.concurrent.CompletableFuture;

public class PaimonOptimizationManager {
    
    private final Catalog catalog;
    private final MetricRegistry metrics;
    
    public PaimonOptimizationManager(String warehousePath) throws Exception {
        // Инициализация каталога и метрик
        Map<String, String> catalogOptions = new HashMap<>();
        catalogOptions.put("warehouse", warehousePath);
        this.catalog = CatalogFactory.createCatalog(catalogOptions);
        this.metrics = new MetricRegistry();
    }
    
    /**
     * Анализ состояния таблицы и выработка рекомендаций по оптимизации
     * 
     * Этот метод собирает статистику по таблице и определяет проблемные места
     */
    public TableHealth analyzeTableHealth(String database, String tableName) throws Exception {
        Table table = catalog.getTable(database + "." + tableName);
        TableHealth health = new TableHealth();
        
        // Собираем базовую статистику
        TableStats stats = collectTableStatistics(table);
        health.stats = stats;
        
        // Анализируем проблемы и даем рекомендации
        analyzeFileDistribution(stats, health);
        analyzeCompactionNeeds(stats, health);
        analyzeQueryPerformance(table, health);
        analyzeBucketSkew(stats, health);
        
        return health;
    }
    
    private TableStats collectTableStatistics(Table table) throws Exception {
        TableStats stats = new TableStats();
        
        // Сканируем манифесты для сбора информации о файлах
        TableScan scan = table.newScan();
        List<DataFileMeta> files = scan.plan().files();
        
        // Группируем файлы по уровням LSM-tree
        Map<Integer, List<DataFileMeta>> filesByLevel = new HashMap<>();
        for (DataFileMeta file : files) {
            filesByLevel.computeIfAbsent(file.level(), k -> new ArrayList<>()).add(file);
        }
        
        // Анализируем каждый уровень
        for (Map.Entry<Integer, List<DataFileMeta>> entry : filesByLevel.entrySet()) {
            int level = entry.getKey();
            List<DataFileMeta> levelFiles = entry.getValue();
            
            LevelStats levelStats = new LevelStats();
            levelStats.level = level;
            levelStats.fileCount = levelFiles.size();
            
            // Вычисляем статистику по размерам файлов
            long totalSize = 0;
            long minSize = Long.MAX_VALUE;
            long maxSize = 0;
            
            for (DataFileMeta file : levelFiles) {
                long size = file.fileSize();
                totalSize += size;
                minSize = Math.min(minSize, size);
                maxSize = Math.max(maxSize, size);
                
                // Считаем фрагментацию (отношение удаленных записей к общим)
                if (file.deleteRecordCount() > 0) {
                    double deleteRatio = (double) file.deleteRecordCount() / 
                                       (file.recordCount() + file.deleteRecordCount());
                    levelStats.fragmentation = Math.max(levelStats.fragmentation, deleteRatio);
                }
            }
            
            levelStats.totalSize = totalSize;
            levelStats.avgFileSize = levelFiles.isEmpty() ? 0 : totalSize / levelFiles.size();
            levelStats.minFileSize = minSize == Long.MAX_VALUE ? 0 : minSize;
            levelStats.maxFileSize = maxSize;
            
            stats.levelStats.put(level, levelStats);
        }
        
        // Общая статистика
        stats.totalFiles = files.size();
        stats.totalSize = stats.levelStats.values().stream()
            .mapToLong(l -> l.totalSize)
            .sum();
        stats.partitionCount = scan.listPartitions().size();
        
        return stats;
    }
    
    private void analyzeFileDistribution(TableStats stats, TableHealth health) {
        // Проверяем распределение файлов по уровням
        // В здоровой LSM-tree каждый следующий уровень должен быть больше предыдущего
        
        int prevLevelFiles = 0;
        for (int level = 0; level <= 5; level++) {
            LevelStats levelStats = stats.levelStats.get(level);
            if (levelStats == null) continue;
            
            if (level == 0) {
                // Level 0 не должен содержать слишком много файлов
                if (levelStats.fileCount > 50) {
                    health.addIssue(
                        HealthIssue.Severity.HIGH,
                        "Too many files in Level 0",
                        String.format("Level 0 has %d files, indicating slow compaction", 
                                    levelStats.fileCount),
                        "Increase compaction threads or reduce write rate"
                    );
                }
                
                // Файлы на Level 0 должны быть небольшими
                if (levelStats.avgFileSize > 256 * 1024 * 1024) { // 256 MB
                    health.addIssue(
                        HealthIssue.Severity.MEDIUM,
                        "Large files in Level 0",
                        "Files in Level 0 are too large, affecting write latency",
                        "Decrease write-buffer-size parameter"
                    );
                }
            } else {
                // Проверяем рост размера файлов между уровнями
                LevelStats prevLevel = stats.levelStats.get(level - 1);
                if (prevLevel != null && levelStats.avgFileSize < prevLevel.avgFileSize * 2) {
                    health.addIssue(
                        HealthIssue.Severity.LOW,
                        "Suboptimal level ratio",
                        String.format("Level %d files are not significantly larger than Level %d", 
                                    level, level - 1),
                        "Consider adjusting compaction.max-file-num"
                    );
                }
            }
            
            prevLevelFiles = levelStats.fileCount;
        }
    }
    
    private void analyzeCompactionNeeds(TableStats stats, TableHealth health) {
        // Анализируем необходимость компактификации
        
        for (LevelStats levelStats : stats.levelStats.values()) {
            // Проверяем фрагментацию
            if (levelStats.fragmentation > 0.3) {
                health.addIssue(
                    HealthIssue.Severity.HIGH,
                    "High fragmentation",
                    String.format("Level %d has %.1f%% deleted records", 
                                levelStats.level, levelStats.fragmentation * 100),
                    "Run full compaction to reclaim space"
                );
            }
            
            // Проверяем количество мелких файлов
            if (levelStats.fileCount > 10 && 
                levelStats.avgFileSize < 64 * 1024 * 1024) { // < 64 MB
                health.addIssue(
                    HealthIssue.Severity.MEDIUM,
                    "Too many small files",
                    String.format("Level %d has %d files with avg size %.1f MB", 
                                levelStats.level, levelStats.fileCount,
                                levelStats.avgFileSize / (1024.0 * 1024.0)),
                    "Increase compaction frequency or trigger manual compaction"
                );
            }
        }
        
        // Проверяем общее количество файлов
        if (stats.totalFiles > 1000) {
            health.addIssue(
                HealthIssue.Severity.HIGH,
                "Excessive file count",
                String.format("Table has %d files total", stats.totalFiles),
                "Enable more aggressive compaction or partition the table"
            );
        }
    }
    
    private void analyzeQueryPerformance(Table table, TableHealth health) {
        // Анализируем потенциальные проблемы с производительностью запросов
        
        // Проверяем настройки кэширования
        Properties tableProps = table.options();
        String cacheSize = tableProps.getProperty("read.cache-size", "0");
        if ("0".equals(cacheSize)) {
            health.addIssue(
                HealthIssue.Severity.LOW,
                "Caching disabled",
                "Read cache is not configured",
                "Set read.cache-size to improve query performance (e.g., '512 MB')"
            );
        }
        
        // Проверяем настройки индексов
        String bloomFilter = tableProps.getProperty("bloom-filter.columns");
        if (bloomFilter == null || bloomFilter.isEmpty()) {
            health.addIssue(
                HealthIssue.Severity.LOW,
                "No bloom filters",
                "Bloom filters not configured for any columns",
                "Add bloom filters for frequently filtered columns"
            );
        }
    }
    
    private void analyzeBucketSkew(TableStats stats, TableHealth health) {
        // Анализируем равномерность распределения данных по bucket'ам
        // Это важно для параллелизма
        
        Map<Integer, Long> bucketSizes = new HashMap<>();
        // Здесь был бы код для анализа распределения по bucket'ам
        // Опущен для краткости
    }
    
    /**
     * Выполнение оптимизации на основе анализа
     */
    public CompletableFuture<OptimizationResult> optimizeTable(
            String database, 
            String tableName,
            OptimizationLevel level) {
            
        return CompletableFuture.supplyAsync(() -> {
            OptimizationResult result = new OptimizationResult();
            
            try {
                Table table = catalog.getTable(database + "." + tableName);
                TableHealth health = analyzeTableHealth(database, tableName);
                
                // Выполняем оптимизацию в зависимости от уровня
                switch (level) {
                    case LIGHT:
                        // Легкая оптимизация: только критические проблемы
                        performLightOptimization(table, health, result);
                        break;
                        
                    case MODERATE:
                        // Умеренная оптимизация: компактификация проблемных уровней
                        performModerateOptimization(table, health, result);
                        break;
                        
                    case FULL:
                        // Полная оптимизация: full compaction + реорганизация
                        performFullOptimization(table, health, result);
                        break;
                }
                
                result.success = true;
                result.message = "Optimization completed successfully";
                
            } catch (Exception e) {
                result.success = false;
                result.message = "Optimization failed: " + e.getMessage();
                result.error = e;
            }
            
            return result;
        });
    }
    
    private void performLightOptimization(Table table, TableHealth health, 
                                         OptimizationResult result) throws Exception {
        // Компактифицируем только Level 0 если там много файлов
        LevelStats level0 = health.stats.levelStats.get(0);
        if (level0 != null && level0.fileCount > 20) {
            compactLevel(table, 0);
            result.actionsPerformed.add("Compacted Level 0");
        }
        
        // Удаляем старые снапшоты
        cleanupSnapshots(table);
        result.actionsPerformed.add("Cleaned up old snapshots");
    }
    
    private void performModerateOptimization(Table table, TableHealth health,
                                           OptimizationResult result) throws Exception {
        // Компактифицируем все уровни с проблемами
        for (HealthIssue issue : health.issues) {
            if (issue.severity == HealthIssue.Severity.HIGH &&
                issue.title.contains("fragmentation")) {
                // Находим уровни с высокой фрагментацией
                for (LevelStats level : health.stats.levelStats.values()) {
                    if (level.fragmentation > 0.3) {
                        compactLevel(table, level.level);
                        result.actionsPerformed.add(
                            String.format("Compacted Level %d (fragmentation: %.1f%%)",
                                        level.level, level.fragmentation * 100)
                        );
                    }
                }
            }
        }
        
        // Также выполняем легкую оптимизацию
        performLightOptimization(table, health, result);
    }
    
    private void performFullOptimization(Table table, TableHealth health,
                                        OptimizationResult result) throws Exception {
        // Full compaction - самая тяжелая операция
        // Перестраивает все файлы для оптимальной структуры
        fullCompaction(table);
        result.actionsPerformed.add("Performed full compaction");
        
        // Пересчитываем статистику
        recomputeStatistics(table);
        result.actionsPerformed.add("Recomputed table statistics");
        
        // Очищаем все ненужные файлы
        cleanupOrphanFiles(table);
        result.actionsPerformed.add("Cleaned up orphan files");
    }
    
    // Вспомогательные методы
    private void compactLevel(Table table, int level) {
        // Запуск компактификации для конкретного уровня
        // Реальная реализация зависит от используемого движка (Flink/Spark)
        System.out.printf("Compacting level %d of table %s%n", level, table.name());
    }
    
    private void fullCompaction(Table table) {
        System.out.printf("Running full compaction for table %s%n", table.name());
    }
    
    private void cleanupSnapshots(Table table) {
        System.out.printf("Cleaning up old snapshots for table %s%n", table.name());
    }
    
    private void recomputeStatistics(Table table) {
        System.out.printf("Recomputing statistics for table %s%n", table.name());
    }
    
    private void cleanupOrphanFiles(Table table) {
        System.out.printf("Cleaning up orphan files for table %s%n", table.name());
    }
    
    private static class TableStats {
        int totalFiles;
        long totalSize;
        int partitionCount;
        Map<Integer, LevelStats> levelStats = new HashMap<>();
    }
    
    private static class LevelStats {
        int level;
        int fileCount;
        long totalSize;
        long avgFileSize;
        long minFileSize;
        long maxFileSize;
        double fragmentation;
    }
    
    private static class OptimizationResult {
        boolean success;
        String message;
        List<String> actionsPerformed = new ArrayList<>();
        Exception error;
    }
    
    private enum OptimizationLevel {
        LIGHT,    // Быстрая оптимизация без остановки записи
        MODERATE, // Умеренная оптимизация с минимальным влиянием
        FULL      // Полная реорганизация (может требовать maintenance window)
    }
    
    private static class HealthIssue {
        enum Severity { LOW, MEDIUM, HIGH }
        Severity severity;
        String title;
        String description;
        String recommendation;
        
        HealthIssue(Severity severity, String title, String description, String recommendation) {
            this.severity = severity;
            this.title = title;
            this.description = description;
            this.recommendation = recommendation;
        }
    }
    
    private static class TableHealth {
        TableStats stats;
        List<HealthIssue> issues = new ArrayList<>();
        
        void addIssue(HealthIssue.Severity severity, String title, 
                     String description, String recommendation) {
            issues.add(new HealthIssue(severity, title, description, recommendation));
        }
    }
    
    // Пример использования
    public static void main(String[] args) throws Exception {
        PaimonOptimizationManager manager = new PaimonOptimizationManager(
            "s3://your-bucket/warehouse"
        );
        
        // Анализируем состояние таблицы
        TableHealth health = manager.analyzeTableHealth("production", "user_events");
        
        // Выводим проблемы
        System.out.println("Table Health Analysis:");
        for (HealthIssue issue : health.issues) {
            System.out.printf("[%s] %s%n", issue.severity, issue.title);
            System.out.printf("  Description: %s%n", issue.description);
            System.out.printf("  Recommendation: %s%n", issue.recommendation);
        }
        
        // Запускаем оптимизацию если есть критические проблемы
        boolean hasCritical = health.issues.stream()
            .anyMatch(i -> i.severity == HealthIssue.Severity.HIGH);
            
        if (hasCritical) {
            System.out.println("Critical issues found, starting optimization...");
            CompletableFuture<OptimizationResult> future = manager.optimizeTable(
                "production", 
                "user_events",
                OptimizationLevel.MODERATE
            );
            
            OptimizationResult result = future.get();
            if (result.success) {
                System.out.println("Optimization completed:");
                result.actionsPerformed.forEach(action -> 
                    System.out.println("  - " + action)
                );
            } else {
                System.err.println("Optimization failed: " + result.message);
            }
        }
    }
}

Этот код демонстрирует комплексный подход к оптимизации Paimon таблиц. Мы анализируем структуру LSM-tree, находим проблемы (слишком много файлов, высокая фрагментация, неоптимальное распределение по уровням) и автоматически применяем оптимизации. Важно понимать, что компактификация - это компромисс между производительностью чтения и нагрузкой на систему. Агрессивная компактификация улучшает скорость запросов, но потребляет ресурсы CPU и I/O.

Заключение: практические выводы и рекомендации

Во второй части мы прошли путь от самой базовой настройки Apache Paimon до сложных сценариев в проде. Давайте подведем итоги и сформулируем ключевые практические выводы.

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

Второе: нужно использовать правильный режим чтения для каждой задачи. Batch режим оптимален для сложной аналитики над большими объемами данных. Streaming режим необходим для real-time мониторинга и continuous обработки. Time travel запросы незаменимы для аудита и отладки. Incremental processing экономит ресурсы при регулярной обработке больших таблиц.

Третье: CDC с версионированием - это мощный паттерн, но требует тщательного проектирования. Решите заранее, нужна ли вам полная история изменений или достаточно текущего состояния. Полная история требует больше места, но дает возможности для глубокого анализа и аудита.

Четвертое: регулярная оптимизация критически важна для поддержания производительности. LSM-tree структура требует периодической компактификации для борьбы с фрагментацией. Автоматизируйте мониторинг и оптимизацию, чтобы проблемы не накапливались.

Наконец, нужно помнить, что Apache Paimon - это не панацея. Это мощный инструмент для определенного класса задач: unified streaming and batch processing с требованиями к низкой задержке и высокой пропускной способности. Если ваша задача попадает в эту категорию, Paimon может существенно упростить архитектуру и снизить операционную сложность.

---

Дополнительные материалы и полезные ссылки:

• Официальная документация: https://paimon.apache.org/docs/1.1/

• Paimon на Python: https://github.com/apache/paimon-python

• Paimon на Rust: https://github.com/apache/paimon-rust

• WebUI для Paimon (проект давно не обновлялся, но UI есть): https://github.com/apache/paimon-webui

• Коннектор для Trino: https://github.com/apache/paimon-trino

• Интересная статья на Хабре про Paimon + StarRocks: https://habr.com/ru/articles/961268/