Пайплайн для Data Lakehouse

~ 20 минут

В этом уроке вернёмся к вопросу о том, как выглядят архитектуры данных. На курсе вы уже выстраивали пайплайн DWH, а в этом уроке рассмотрите гибрид DWH и Data Lake — Data Lakehouse. Так вы увидите на практике разделение ответственности по работе со структурированной и полуструктурированной и неструктурированной информацией.

Часто по пути к Data Lakehouse компании уже имеют два решения: одно для оперативной отчётности, а другое ― для накопления данных и применения машинного обучения (возможно, уже даже в состоянии «болота»). Например, в финтехе в процессе KYC и для анализа рисков изучается вся доступная информация: соцсети, активность в интернете, связь с другими людьми, фотографии и прочее.

Поддерживать два решения дорого и неудобно в контексте масштабируемости, особенно в части DWH, поэтому Data Lakehouse становится эволюционным решением.

В этом уроке вы разберётесь, из чего состоит пайплайн Data Lakehouse и как работает эта архитектура. А в следующем мы дадим инструкцию, чтобы собрать пайплайн на практике.

Итак, Data Lakehouse состоит из двух основных частей:

Хранилище — масштабируемое и гибкое решение для работы со структурированными (например, csv), частично структурированными (например, json) и неструктурированными данными (например, текстовый файл).
Уровень метаданных — единый «каталог», который предоставляет метаданные для всех объектов в хранилище и позволяет управлять данными (транзакции, версионирование, управление схемой).

Поговорим про технологии, которые будут использоваться для построения Data Lakehouse. Начнём с хранилища — в Data Lakehouse на этом уровне хранятся «сырые» данные-объекты, которые затем могут быть подвергнуты обработке.

Хранилище и его типы

Есть три основных типа хранения информации: файловый, блочный и объектный.

Файловые хранилища (file storage) — это привычные файловые системы (FAT32, NTFS, EXT4 и другие), которые хранят информацию в иерархической структуре. Для небольших объёмов и информации, которая ложится в древовидную структуру, они работают хорошо, но на больших объёмах (терабайтах) и при интенсивных операциях чтения или записи производительность падает.
Блочные хранилища (block storage) — данные делятся на блоки одинакового размера, что позволяет достигать хорошей производительности даже под нагрузкой. Но есть два основных недостатка: высокая стоимость и слабая поддержка работы с метаданными — их приходится обрабатывать на уровне приложения.
Объектные хранилища (оbject storage) — это тип хранения, при котором информация разбивается на объекты, каждый из которых сохраняется отдельно, имеет собственный идентификатор и метаданные. Доступ к объекту происходит по ключу-идентификатору. Хранилище с таким типом хранения хорошо масштабируется, держит нагрузку, поддерживает механизмы безопасности. Оно очень просто в управлении. Пожалуй, самым популярным вариантом реализации объектного хранения являются так называемые S3-совместимые хранилища.

S3-совместимое хранилище

Amazon Simple Storage Service (S3) — изначально это был сервис от Amazon Web Services со своими протоколами и спецификациями для хранения данных и получения их через REST API. Но решение оказалось настолько удобное, что появились аналоги, которые реализуют спецификацию, и они уже называются S3-совместимое хранилище.

Они построены с использованием объектного типа хранения, имеют Restful API, хорошо масштабируются и держат нагрузку. Они S3-совместимые, и поэтому вендоронезависимые и решения на базе такого хранилища легко мигрировать на другую технологию или к другому провайдеру в случае облачной инсталляции.

Примеры хранилищ, которые реализуют спецификацию

Yandex Object Storage ― облачная реализация от Яндекс Облака
MinIO ― популярная легковесная реализация
Zenko CloudServer ― реализация от Zenko, полностью совместимая со спецификацией, но на текущий момент наблюдается застой в развитии продукта
LeoFS ― быстрое, распределённое и на Erlang/OTP
Ceph ― «тяжёлое» и многофункциональное решение от RedHat, часто можно встретить в энтерпрайз-среде

При выборе следует учитывать требования к хранению. Например, Ceph не только объектное хранилище, но и поддерживает и блочное, и файловое.

В нашем случае «комбайн» не нужен, но ситуации бывают разные: например, если у нас уже есть хранилище на Ceph или политика компании не позволяет использовать ничего кроме, ну или у нас просто некое легаси, которое требует именно его. Для нашего пайплайна воспользуемся MinIO как решением, которое полностью реализует спецификацию S3 и отличается легкостью развёртывания и настройки.

Уровень метаданных

С уровнем метаданных всё немного сложнее. Этот слой как раз «даёт» возможность совершать ACID-транзакции, которых не хватает в Data Lake, и обеспечивать хорошую масштабируемость.

Уровень метаданных реализуется с помощью так называемых табличных форматов — Table format, где данные в объектном хранилище структурируются и унифицируются в виде единой таблицы. Благодаря тому, как метаданные организованы и управляются, можно ответить на вопрос «Какие данные есть в таблице?». Такой формат — некая «умная» надстройка над «глупым» хранилищем всего подряд.

Форматы хранения метаданных

Основных реализаций хранения три:

Apache Iceberg (Netflix): Iceberg — это формат данных, который помогает управлять большими таблицами в озёрах данных. Он поддерживает транзакции, версионность и работает с разными типами файлов (например, Parquet и ORC). Iceberg делает запросы к данным быстрыми и упрощает работу с большими объёмами.
Apache Hudi (Uber): Hudi — формат данных, который позволяет обновлять и удалять записи. Он хорошо подходит для работы с изменяющимися данными в реальном времени. Hudi поддерживает упорядочивание данных и помогает держать их в актуальном состоянии.
Delta Lake (Databricks): Delta Lake делает хранение данных более надёжным. Он добавляет транзакции, отслеживает версии данных и помогает исправлять ошибки. Delta Lake тесно связан с Apache Spark и позволяет быстро обрабатывать данные для аналитики.

	Iceberg	Hudi	Delta Lake
ACID-Транзакции	✅	✅	✅
Эволюция партиций	✅	✅	❌
Эволюция схемы	✅	✅	✅
Путешествие во времени	✅	✅	✅
Управление проектом	Apache Project with a diverse PMC (top-level project)	Apache Project with a diverse PMC (top-level project)	Linux Foundation Project with an all-Databricks TSC
Вклад сообщества	347 участников сообщества, 3836 замёрженных пул-реквестов, 465 открытых пул-реквестов	379 участников сообщества, 4496 замёрженных пул-реквестов, 393 открытых пул-реквестов	231 участников сообщества, 26 замёрженных пул-реквестов, 73 открытых пул-реквестов
Совместимость инструментов для чтения	Apache Hive, Dremio Sonar, Apache Flink, Apache Spark, Presto, Trino, Athena, Snowflake, Databricks Spark, Apache Impala, Apache Drill, BigQuery	Apache Hive, Apache Flink, Apache Spark, Presto, Trino, Athena, Databricks Spark, Redshift, Apache Impala, BigQuery	Apache Hive, Dremio Sonar, Apache Flink, Databricks Spark, Apache Spark, Databricks SQL Analytics, Trino, Presto, Snowflake, Redshift, Apache Beam, Athena
Совместимость инструментов для записи	Apache Hive, Dremio Sonar, Apache Flink, Apache Spark, Trino, Athena, Databricks Spark, Debezium, Kafka Connect, BigQuery, Snowflake	Apache Flink, Apache Spark, Databricks Spark, Debezium, Kafka Connect	OSS Delta Lake: Trino, Apache Spark, Databricks Spark, Apache Flink, Debezium, AthenaDatabricks Delta Lake: Databricks Spark, Kafka Connect
Поддержка форматов файлов	Parquet, ORC, AVRO	Parquet, ORC	Parquet

И небольшой бенчмарк от Delta Lake в различных персентилях:

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

Apache Iceberg

Архитектура Apache Iceberg разделена на те же два уровня Data Lakehouse:

Уровень данных — Apache Iceberg работает с открытыми файловыми форматами (Open File Formats), такими как Parquet, Avro и ORC. Это означает, что сами данные хранятся в широко используемых форматах файлов;
Слой метаданных — Iceberg добавляет уровень управления таблицами (Open Table Formats), который отслеживает состояние таблиц, версии данных, транзакции и упрощает доступ к данным.

Сверху над айсбергом располагается каталог — он решает проблему поиска данных из разных источников, используя метаданные для идентификации наборов данных. Каталог хранит указатель на актуальный файл метаданных, который можно использовать для записи и чтения (например, Nessie).

Iceberg не является механизмом хранения, механизмом выполнения или сервисом.

Спецификация формата таблицы.
Набор API и библиотек для взаимодействия с таблицами в соответствии с этой спецификацией.

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

Эволюция схемы — можно легко изменять структуру;
Сохранение состояния данных в определённый момент времени (снепшоты) и возврат к этим состояниям (роллбэк данных);
Разделение данных на части с группировкой похожих строк при записи. Например, он может разделять события журнала по дате и группировать их в файлы с одинаковой датой события. Так он может пропускать файлы с другими датами, в которых нет полезных данных, и ускорять выполнение запросов;
Скрытое партиционирование, то есть система автоматически управляет партиционированием данных, не требуя от пользователя явно указывать партиционные ключи при запросах.

Как работает Iceberg?

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

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

Пока таблица пустая, манифестов нет.

После вставки данных в таблицу (обычной командой INSERT через каталог) появляется список манифестов и конкретный манифест для каждой вставки, а также создаётся файл в объектном хранилище.

Если обновить эту таблицу, обновление создаст новый файл метаданных со снимком s2, а s1 станет уже неактуальным.

💡 Файлы данных в моментальных снимках отслеживаются с помощью одного или нескольких файлов-манифестов, которые содержат строку для каждого файла данных в таблице, данные раздела файла и его показатели.

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

Рекомендуется периодически архивировать или удалять неактуальные снимки. Например, снимки с потерянными файлами метаданных, которые не были зафиксированы во время записи, можно обработать повторно или удалить, используя метод API. В противном случае будет расти количество мусора.

Каталоги

Самое время определиться с каталогом, чтобы в следующем уроке собрать весь пайплайн целиком.

Есть две категории каталогов — основанные на файлах и на межсервисном взаимодействии.

Каталоги, основанные на файлах

Таким каталогом является, например, Hadoop Catalog.

Каталог содержит файл version-hint.text, в котором указана последняя версия файла метаданных (metadata.json). Когда система выполняет запросы, она сначала проверяет этот файл, чтобы найти путь к актуальным метаданным.

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

Каталоги для межсервисного взаимодействия

Эти каталоги используют сервисы для хранения и управления метаданными.

JDBC метаданные хранятся в реляционной базе данных через JDBC. Это упрощает управление в системах с классическими базами данных.
Hive Metastore использует метастор Hive для хранения информации о таблицах. Подходит для экосистемы Hadoop и Spark.
Nessie работает через REST API, что делает его простым и универсальным для интеграции. Поддерживает версионирование данных.
Glue — каталог от AWS, который предоставляет централизованное хранилище метаданных и легко интегрируется с другими сервисами AWS.

В следующем уроке мы перейдём к практике, где будем использовать Nessie — каталог, который взаимодействует с Iceberg по REST API и даёт возможности наподобие Git — создавать ветки данных, коммитить их и мёржить.

Для получения пользовательского интерфейса воспользуемся Dremio — платформой «данные как услуга». Инструмент может подключаться к различным источникам, включая базы данных, озёра данных и облачные платформы хранения. Затем данные можно запросить с помощью SQL.

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

Задание 1

В ходе работы с Data Lakehouse через Nessie возникла необходимость создать ветку для тестирования данных перед внесением изменений в основную ветку.

Какая последовательность SQL-команд позволит создать новую ветку, обновить данные и проверить, что изменения не затронули основную ветку?

SQL

CREATE BRANCH test_branch;
UPDATE orders SET total=-1 WHERE id=1;
USE BRANCH test_branch;
SELECT * FROM orders;

Ветка создаётся, но команда USE BRANCH выполняется после внесения изменений. Изменения применятся не в новой ветке, а в текущей ветке (обычно main).

SQL

USE BRANCH test_branch;
CREATE BRANCH test_branch;
UPDATE orders SET total=-1 WHERE id=1;
SELECT * FROM orders;

Ошибка. Невозможно переключиться на ветку, которая ещё не существует. Команда USE BRANCH должна выполняться после создания ветки с помощью CREATE BRANCH.

SQL

CREATE BRANCH test_branch IN catalog;
USE BRANCH test_branch;
UPDATE orders SET total=-1 WHERE id=1;
SELECT * FROM orders;

Создание ветки в Nessie происходит с использованием CREATE BRANCH с указанием имени ветки и IN catalog. Затем переключение на ветку выполняется через USE BRANCH.

SQL

CREATE BRANCH test_branch IN main;
UPDATE orders SET total=-1 WHERE id=1;
USE BRANCH test_branch;
SELECT * FROM orders;

Команда CREATE BRANCH не требует указания IN main. Правильный синтаксис — CREATE BRANCH test_branch IN catalog. Кроме того, изменения применяются до переключения ветки, что нарушает изоляцию.

Задание 2

Компания "DataFlow Analytics" использует Apache Iceberg для управления большими объёмами аналитических данных в Data Lakehouse. Команда сталкивается со следующими задачами:

Чтобы ускорить выполнение запросов, команда хочет настроить скрытое партиционирование данных по дате заказа. Как Iceberg автоматически управляет партиционированием?

Явно указывая партиционные ключи при каждом запросе

Apache Iceberg позволяет скрытое партиционирование, которое управляет партициями автоматически. Пользователю не нужно указывать партиционные ключи при каждом запросе, что уменьшает сложность работы с таблицами.

Автоматически группируя похожие строки при вставке данных

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

Создавая отдельные таблицы для каждой партиции

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

Используя внешние партиционные файлы

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

Задание 3

Компания "BigData Insights" разрабатывает пайплайн для Data Lakehouse, используя такие инструменты, как MinIO для хранения, Nessie для управления версиями данных и Apache Iceberg для контроля таблиц и метаданных. Ваша задача — помочь команде инженеров построить и проверить пайплайн данных для эффективной аналитики.

Инженеры настраивают пайплайн для передачи и хранения данных. Какой порядок действий правильный?

Настройка Airflow для передачи данных в хранилище.

Шаг 3

Airflow не может корректно функционировать до настройки хранилища (MinIO) и системы метаданных (Nessie).

Запуск MinIO для S3-совместимого хранилища.

Шаг 1

MinIO запускается первым, так как оно является базовым хранилищем.

Подключение Nessie как каталога метаданных.

Шаг 2

Nessie подключается вторым для управления метаданными.

Итоги

Data Lakehouse объединяет два ключевых уровня: хранилище и метаданные. Хранилище служит для хранения любых типов данных, включая структурированные, частично структурированные и неструктурированные. Уровень метаданных действует как единый каталог, который предоставляет информацию о данных, поддерживает транзакции, версионирование и управление схемами.
В Data Lakehouse используются три типа хранилищ: файловые, блочные и объектные. Наиболее популярные — объектные хранилища S3. Они позволяют эффективно хранить «сырые» данные-объекты, которые позже обрабатываются для анализа и легко масштабируются.
Apache Iceberg управляет таблицами данных, создавая файлы метаданных, которые хранят схему таблицы, партиции, версии и ссылки на данные. Когда таблица создаётся или изменяется, Iceberg обновляет метаданные и таким образом обеспечивает согласованность и упрощает обработку данных. Iceberg считается «золотой серединой» между сложными корпоративными Hudi и Delta Lake, которые для некоторых функций используют платные улучшения.