Как мы переехали с Greenplum на Delta Table

У нас было 2 кластера Clickhouse, 1 кластер GreenPlum’a, 1 кластер Kubernetes’a, DataVault 2.0, гора dbt витрин и всего такого, а еще Dagster. Не то чтобы это все было нужно в архитектуре, но раз начал коллекционировать сервисы, то иди в своем увлечении до конца. Единственное, что нас беспокоило — это стоимость. 

Решение о полной переделке архитектуры пришло внезапно, как галлюцинация в ночной пустыне — мы поняли, что больше нельзя ждать. Теперь наши данные обрабатываются быстрее, чем мысли в голове на ЛСД, и мы можем персонализировать клиентский опыт так, что он становится почти реальным. 

---

Всем привет! На связи Артем, и если вы готовы погрузиться в мир цифрового безумия и увидеть, как мы превратили хаос в порядок, то держитесь крепче и читайте дальше.

Когда-то мы уже писали о том, как построить DataVault на Greenplum, со статьей можно ознакомиться здесь. Кратко расскажем о причинах, подтолкнувших нас пересмотреть подход — это оптимизация расходов и отсутствие экспертов в администрировании GreenPlum.

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

Выбор нового решения

При выборе нового решения мы учитывали несколько ограничений:

• Dagster: для управления задачами

• Kubernetes: для оркестрации контейнеров

• Yandex Cloud: как основная облачная платформа

Основные компоненты нового решения:

• Yandex Object Storage: для хранения

• Delta Table: для управления

• Apache Spark: для обработки

Почему именно эти технологии?

Переход с архитектуры Data Vault 2.0 на архитектуру Delta Lake (S3 + Delta Table) может предложить ряд существенных преимуществ:

• Транзакционная целостность и ACID-свойства — Delta Lake обеспечивает надежность и целостность данных при параллельных операциях

• Производительность и оптимизация запросов — структура Data Vault 2.0 может приводить к сложным и менее оптимизированным запросам, требующим значительных усилий для оптимизации

• Гибкость и масштабируемость — Data Vault 2.0 хорошо подходит для интеграции из  различных источников, но может потребовать значительных ресурсов для масштабирования под высокие нагрузки, тогда как Delta Lake спроектирован для работы с большими объемами данных в облачной среде и легко масштабируется в зависимости от нагрузки без серьезных затрат

• Упрощенное управление схемой данных — delta table поддерживают автоматическую эволюцию схемы и принудительное соблюдение схемы, что упрощает внесение изменений в структуру данных и управление ими

• Экосистема и интеграция с другими инструментами — delta table легко интегрируется с apache spark, что упрощает разработку и эксплуатацию аналитических приложений

Apache Spark

Одним из преимуществ Apache Spark является возможность его локального развертывания, что особенно удобно для разработки и тестирования. На текущий момент обработка наших данных происходит именно таким образом.

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

В самом простом виде нашу архитектуру можно представить следующим образом:

Данные хранятся в Yandex Object Storage в формате Delta Table. Это позволяет нам воспользоваться преимуществами версионирования данных и транзакционной целостности. Помимо этого, мы используем S3 для хранения артефактов ML моделей.

В работе с форматом Delta Table правильное распределение данных по партициям играет ключевую роль. Учитывая разнообразие наших источников данных, мы внедрили партиционирование не только по дате, но и по источнику. Это значительно повышает скорость и эффективность чтения данных. При грамотном выборе партиции Spark применяет partition pruning, избегая необходимости читать все файлы в таблице, что оптимизирует производительность системы.

Для нашего MVP мы выбрали использование кластера Spark на одном узле, настроив SparkSession следующим образом:

SparkSession.builder.appName().master(f"local[*]")

Полная конфигурация Spark’a для работы с S3 может выглядеть следующим образом:

from delta import configure_spark_with_delta_pip from pyspark.sql import SparkSession  packages = [     "io.delta:delta-core_2.12:2.0.0",     "org.apache.hadoop:hadoop-aws:3.3.1", ]  builder = (SparkSession.builder.appName().master(f"local[*]")     .config("spark.driver.memory", "16g")     .config("spark.hadoop.fs.s3a.bucket.<bucket_name>.access.key", "<backet_access_key>")     .config("spark.hadoop.fs.s3a.bucket.<bucket_name>.secret.key", "<backet_secret_key>")     .config("spark.hadoop.fs.s3a.bucket.<bucket_name>.endpoint", "<backet_endpoint>")     .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")     .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")     .config("spark.hadoop.fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")     .config("spark.hadoop.fs.s3a.fast.upload", "true")     .config("spark.sql.ui.explainMode", "extended")     .config("spark.sql.parquet.datetimeRebaseModeInWrite", "CORRECTED")     .config("spark.sql.parquet.int96RebaseModeInWrite", "CORRECTED")     .config("spark.sql.sources.partitionOverwriteMode", "dynamic")     .config("spark.driver.extraJavaOptions", _spark_with_newer_jvm_compatibility_options) )  spark = builder.getOrCreate() spark.sparkContext.setLogLevel("ERROR")

Такой подход позволяет запускать Spark локально на той же машине, где создается сессия. Благодаря этому, мы можем запускать Spark непосредственно в подах Dagster и удобно отлаживать запросы локально.

В работе с Dagster мы используем ассеты и менеджеры. Вся бизнес-логика размещается внутри ассета, а управление вводом-выводом передается менеджеру. Таким образом, нам было достаточно написать Spark менеджер и адаптировать запросы в коде для работы со Spark. 

Для экономии ресурсов во время простоя мы внедрили автоскейл группы прерываемых узлов в Managed Service for Kubernetes от Яндекс Облака. Развернули с помощью Terraform, используя готовый модуль. Мы настроили группы так, чтобы они могли масштабироваться до нуля узлов при отсутствии нагрузки. Для подов Dagster у нас выделена отдельная группа узлов, где всегда работает хотя бы один узел. Таким образом, система практически не потребляет ресурсы в периоды простоя.

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

@job(         tags={         "dagster-k8s/config": {             "container_config": {                 "resources": {                     "requests": {"cpu": "250m", "memory": "64Mi"},                     "limits": {"cpu": "500m", "memory": "2560Mi"},                 },             },     }, ) def my_job():     my_op()  Либо, при использовании ассетов: my_job = define_asset_job( name=’my_job’, selection=AssetSelection.assets(my_asset), tags={         "dagster-k8s/config": {             "container_config": {                 "resources": {                     "requests": {"cpu": "250m", "memory": "64Mi"},                     "limits": {"cpu": "500m", "memory": "2560Mi"},                 },             },     }, )

Для записи данных в Clickhouse мы используем библиотеку clickhouse-connect, так как наш исходный код уже был написан с ее помощью. Однако стоит отметить, что для Spark есть драйвер для работы с Clickhouse, позволяя использовать Spark SQL для выполнения запросов и обработки данных в Clickhouse.

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

Важное замечание: Spark + Delta Table по умолчанию поддерживают механизм оптимистичных блокировок для write-запросов, выполняемых в одном Spark кластере, но не для write-запросов из разных Spark кластеров. Чтобы включить такую поддержку, необходимо добавить в конфигурацию DynamoDB. В Яндекс Облаке YDB реализует интерфейс DynamoDB, но его интеграция оказалась для нас сложной задачей. Поэтому мы решили использовать механизм пессимистичных блокировок, встроенный в Dagster, через функцию tag_concurrency_limits. Все джобы, записывающие данные в таблицу, мы помечаем тегом с названием этой таблицы. Dagster, в свою очередь, предотвращает одновременное выполнение таких джоб, что обеспечивает корректное управление блокировками и исключает конфликтные write-запросы. 

Таким образом, конфигурация вашего dagsterDaemon будет выглядеть как-то так:

dagsterDaemon:  enabled: true    image:    repository: "docker.io/dagster/dagster-celery-k8s"    tag: ~    pullPolicy: Always    heartbeatTolerance: 300    runCoordinator:    enabled: true    type: QueuedRunCoordinator    config:      queuedRunCoordinator:        maxConcurrentRuns: 10        tagConcurrencyLimits:          - key: "single-thread"            value:              applyLimitPerUniqueValue: true            limit: 1

После чего в настройки вашего задания можно добавить:

my_job = define_asset_job( name=’my_job’, selection=AssetSelection.assets(my_asset), tags={         “single-thread”: “my_job”,         "dagster-k8s/config": {             "container_config": {                 "resources": {                     "requests": {"cpu": "250m", "memory": "64Mi"},                     "limits": {"cpu": "500m", "memory": "2560Mi"},                 },             },     }, )

Преимущества нового подхода:

• Экономия расходов

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

• Гибкость и масштабируемость

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

• Поддержка существующих процессов

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

Заключение

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

А в данный момент мы подготавливаем инфраструктуру для внедрения горизонтально масштабируемого Spark’a и поддержки Hive Metastore, о чем постараемся рассказать в следующей статье.

---

Ну вот и все!) Благодарю за внимание и буду рад услышать ваше мнение) Поделитесь своим опытом, задавайте вопросы и расскажите, как вы справляетесь с подобными задачами)