Lake Berryessa by chateaugrief

Выбор формата хранения данных при создании Data Lake влияет на производительность озера, удобство его использования и совместимость с другими инструментами. 

Команда разработки Cloud Big Data от VK Cloud Solution перевела статью — сравнение трех форматов данных: Apache Hudi, Apache Iceberg и Delta Lake. Посмотрим, какие в них применяются подходы к производительности обновлений, параллелизму и совместимости, и разберем, какой формат больше подойдет под ваши задачи.

Почему мы рассматриваем именно эти три формата

Apache Hudi, Apache Iceberg и Delta Lake — лучшие в своем классе форматы, разработанные специально для озер данных. Все они решают три задачи:

1. Целостность данных. Атомарные транзакции гарантируют, что операции обновления и добавления в озере не завершатся сбоем на полпути. Это позволяет избежать повреждения данных.
   
2. Согласованность обновлений. Предотвращают сбой чтения или возврат неполных результатов во время записи. Решают проблему возможной одновременной записью.
   
3. Масштабируемость данных и метаданных. Устраняют узкие места с API объектного хранилища и соответствующими метаданными, когда таблицы увеличиваются до тысяч партиций и миллиардов файлов.
   

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

Платформенная совместимость

Hudi. Uber изначально разрабатывал Hudi как решение с открытым кодом для инкрементных обновлений колоночно-ориентированных форматов данных. Он поддерживает:

• прием данных из разных источников, в основном Apache Spark и Apache Flink;
  
• чтение данных из внешних источников, таких как Apache Kafka. Работает за счет внешней утилиты на основе Spark;
  
• чтение данных из Apache Hive, Apache Impala и PrestoDB; 
  
• синхронизацию табличной схемы с Hive Metastore;
  

Iceberg. Netflix разрабатывал этот формат, чтобы решить проблемы с производительностью, масштабируемостью и управляемостью, которые возникали при хранении больших наборов данных в партиционированных таблицах Hive на S3.

Iceberg поддерживает:

• чтение и запись данных на основе Apache Spark, в том числе структурированный стриминг Spark;
  
• операции чтения в Trino (PrestoSQL) и, с некоторыми ограничениями, операции удаления данных;
  
• чтение и запись в Apache Flink;
  
• чтение данных Apache Hive.
  

Delta Lake. Его разрабатывает команда Databricks, создатели Apache Spark, как проект с открытым исходным кодом. Он поддерживает:

• глубокую интеграцию операций записи и чтения со Spark;
  
• чтение данных для Presto, AWS Athena, AWS Redshift Spectrum и Snowflake с использованием SymlinkTextInputFormat в Hive. Хотя для этого нужно экспортировать файл symlink.txt для каждого раздела таблицы Delta, а это довольно дорогое удовольствие для больших таблиц.
  

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

Есть несколько способов реализовать поддержку обновлений больших неизменяемых объектов на уровне строк. И в каждом случае придется идти на компромисс в производительности и скорости. Посмотрим, как эти форматы выполняют операцию UPSERT, а также затронем дополнительную оптимизацию, связанную с производительностью чтения.

Hudi. Таблицы Hudi в зависимости от типа предлагают два варианта выполнения операции UPSERT с разной производительностью:

• Copy on Write Table. Обновления записываются только в колоночные файлы Parquet, создавая новые объекты. Затраты на запись увеличиваются, но до нуля сокращается дополнительная нагрузка при чтении. Это решение идеально для рабочих нагрузок, сопряженных с интенсивными операциями чтения.
  
• Merge on Read Table. Обновления сразу попадают в Row-based лог-файлы, которые периодически преобразуются в Parquet. Запросы могут содержать последние данные файла журнала — с помощью «переключателя» пользователи выбирают между латентностью данных и эффективностью выполнения запросов.
  

С подробной информацией о настраиваемых параметрах производительности в Hudi можно ознакомиться в статье «Задержки производительности для записи данных в Hudi».

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



Iceberg. С выпуском версии Spark 3.0, которая увидела свет летом 2020-го, Iceberg стал поддерживать операции UPSERT через запросы MERGE INTO. Они реализуются прямолинейным подходом Copy-on-write: файлы с записями, которые нужно обновить, сразу же перезаписываются.

Однозначное преимущество Iceberg — производительность чтения таблиц со множеством партиций. Iceberg работает с файлами манифеста, которые сопоставляют объект с партициями и ведут статистику на уровне столбцов. Поэтому отпадает необходимость в дорогостоящем составлении списков директорий хранения объектов или поиске данных партиции в Hive.

Кроме того, манифесты Iceberg позволяют присваивать один и тот же файл нескольким разделам одновременно. Таким образом, таблицы Iceberg эффективно обеспечивают Partition Pruning и оптимизацию задержки выборочных запросов.

Delta Lake. Во время операции MERGE Delta использует пропуск данных Metadata-informed, чтобы понять, нужно ли вставить, обновить или удалить данные в тех или иных файлах. Потом она выполняет эти операции и записывает их как коммит в файл журнала JSON, который называется Delta Log. Эти лог-файлы перезаписываются один раз на десять коммитов как файлы контрольной точки Parquet, в которых сохраняется всё состояние таблицы. Это делает ненужной операцию чтения всех файлов журнала при доступе к данным.

Чтобы сохранить высокую производительность, таблицам Delta нужно периодически проходить процедуру сжатия (compaction): много маленьких файлов Parquet объединяют в небольшое количество больших файлов (оптимально размером около 1 ГБ, но хотя бы 128 МБ). В корпоративной версии Databricks Delta Engine поддерживается автосжатие, когда этот процесс запускается автоматически, а также другие скрытые варианты оптимизации записи данных.

Delta Engine обгоняет по производительности своего собрата с открытым исходным кодом за счет индексирования ключей с использованием фильтров Блума, Z-Ordering для улучшенного File Pruning во время чтения, локального кэширования и других функций. 

Гарантированное выполнение параллельных процессов

Что будет, если кто-то начнет читать таблицу во время обновления? А если несколько операций записи попытаются внести в таблицу конфликтующие изменения? Есть два способа решения этой проблемы:

• Multiversion Concurrency Control, MVCC. Используется журнал логических транзакций, в который добавляют все изменения.
  
• Optimistic Concurrency Control, OCC. С его помощью можно одновременно выполнять несколько операций записи, просто проверяя наличие конфликтов перед окончательным коммитом. При обнаружении конфликта одна из операций повторяется до тех пор, пока не будет выполнена.
  

Hudi. Здесь используются и MVCC, и OCC. MVCC в Hudi означает, что все операции записи должны быть упорядочены в центральном журнале. Чтобы гарантировать выполнение параллельных процессов, Hudi не разрешает больше одного процесса параллельной записи. То есть в каждый отдельный момент времени в таблицу может выполняться только одна операция записи.

Чтобы преодолеть это ограничение, сейчас Hudi в экспериментальном режиме предлагает и OCC. Для реализации этой функции нужно, чтобы в Apache Zookeeper или Hive Metastore блокировались и изолировались отдельные файлы.

Iceberg. Таблицы Iceberg поддерживают ОСС с помощью операции Atomic Swapping с файлами метаданных во время обновления.

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

1. При каждой записи создается новый «снимок» таблицы.
   
2. Затем операция записи пытается выполнить операцию Compare-And-Swap (CAS) с идентификатором текущего снимка.
   
3. Операция завершается успешно, если не появляется нового «снимка» таблицы с новым идентификатором.
   
4. Если же в это время еще одна операция записи делает коммит, то создается новый «снимок», и первая операция записи будет вынуждена повторять попытки, пока не выполнится успешно.
   

В распределенных системах, таких как HDFS, это можно сделать встроенными средствами. Для S3 требуется дополнительный компонент для хранения указателей — на данный момент поддерживается только Hive Metastore.

Delta Lake. В документации Delta говорится, что она применяет ООС. Большая часть операций в озере добавляет данные в партицию, где они упорядочены по времени, а значит, конфликтов не возникает.

Если два процесса одновременно добавляют коммиты в файл журнала, Delta проверяет, не пересекаются ли изменения, чтобы по возможности оба были выполнены успешно.

Но если несколько операций начнут перезаписывать сделанные в журнале записи друг друга, объектному хранилищу нужно как-то обеспечить выполнение операции CAS, либо произойдет ошибка записи. Как и в случае с Iceberg, эта возможность поддерживается в готовой версии на основе HDFS, но не поддерживается для S3. В результате в Delta на AWS не поддерживаются операции записи из нескольких кластеров Spark с гарантиями выполнения транзакции.

Примечание: Корпоративная версия Delta Engine поддерживает запись с нескольких кластеров на S3 с внешнего сервера синхронизации, управляемого самой Databricks.

Какое решение подходит для вас

Мы выяснили, в чем сходства и различия между Apache Hudi, Delta Lake и Apache Iceberg. Настало время решить, какой формат лучше всего подходит в вашем случае. Мои рекомендации зависят от сценария, который вы используете. 

Выбирайте Iceberg, если ваша основная задача — не изменять имеющиеся записи, а управлять в хранилище огромными таблицами с метаданными с более чем с 10 тысячами партиций. Переход на Iceberg избавит вас от проблем с производительностью, связанных с листингом объектов S3 или проблемой чтения большого числа партиций в Hive Metastore.

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

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

Выбирайте Delta Lake, если вы в основном используете Spark и ожидаете относительно небольшой объем операций записи. Если вы уже клиент Databricks, Delta Engine позволит серьезно повысить производительность операций чтения и записи, в том числе при параллельных процессах. Так что имеет смысл и дальше пользоваться преимуществами их экосистемы.

Что касается других дистрибутивов Apache Spark, Delta Lake как решение с открытым исходным кодом, скорее всего, всегда будет отставать от Delta Engine.

Команда VK Cloud Solutions развивает собственные Big-Data-решения. Будем признательны, если вы их протестируете и дадите обратную связь. Для тестирования пользователям при регистрации начисляем 3000 бонусных рублей.

Что почитать по теме:

1. Как и зачем разворачивать приложение на Apache Spark в Kubernetes.
   
2. Форматы файлов в больших данных: краткий ликбез.