Блог Eleven Labs — Создание платформы данных, обратная связь (REX)

Контекст REX вокруг создания платформы данных для нашего клиента

В рамках миссии Data Engineer для клиента компании Студия Одиннадцать ЛабораторийЯ присоединился к подразделению «Фабрика данных», чтобы анализировать и понимать поведение пользователей. Это помогает лучше направлять добавление функций и продуктов, которые будут запущены.

Группа обработки данных реализовала Poc («Подтверждение концепции»). Он построен на основе конвейера ELT (извлечение, загрузка, преобразование) с использованием следующих технологий: Google Cloud Platform, Talend, dbt и Power BI.

Чтобы быстро протестировать PoC, конвейер запускается на Jenkins. Однако отказ от использования Jenkins для выполнения конвейеров не остался без последствий. Дженкинс для этой работы не подходит: никаких повторов, написание конвейера на Groovy, только одна среда исполнения.

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

Именно в этом контексте начинается моя миссия.

Конвейер ELT: извлечение, загрузка, преобразование

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

Как работает этот трубопровод? Каковы дальнейшие действия? Каковы потребности этого трубопровода?

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

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

Извлечение и загрузка данных в Google BigQuery

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

В источниках данных у меня есть:

Для этапа экстракции и загрузки в Google BigQueryTalend был создан для выполнения этой работы. Это инструмент, который позволяет создавать полные конвейеры ETL (извлечение, преобразование, загрузка; не путать с ELT). Здесь он использовался только для выполнения фаз извлечения и загрузки в Google BigQuery.

Для разработки и модификации требуется толстый клиент и ручная компиляция конвейера извлечения.

Как только данные попадут в BigQuery, можно начать этап преобразования.

Трансформация с помощью dbt

Второй шаг — преобразование данных.

Это преобразование осуществляется с помощью дбт непосредственно в хранилище данных Google BigQuery.

dbt позволяет организовать преобразование данных и шаблонизировать SQL-запросы. Google BigQuery выполняет SQL-запросы.

На этом этапе преобразования создаются новые структуры данных для хранения результатов вычислений. Эти агрегаты данных затем отображаются с помощью инструмента визуализации данных: Power BI.

Просмотр данных с помощью Power BI

Наконец, на последнем этапе этого конвейера происходит отображение данных.

Конечная цель всей этой работы — избежать выполнения всех вычислений при отображении аналитических отчетов. Без предшествующей работы по расчетам и агрегированию данных отображение графиков заняло бы очень много времени.

Этот конвейер функционален и уже работает в Jenkins. Давайте посмотрим на архитектуру новой платформы данных.

Архитектура платформы данных

В предыдущей части мы увидели, как работает конвейер ELT в Дженкинсе. Цель состоит в том, чтобы перенести это на более надежную платформу, адаптированную для такого типа работы.

Для этого нам нужен инструмент для координации этих различных этапов конвейера и их перезапуска в случае ошибки. Apache Airflow — идеальный кандидат. Google предлагает управляемую версию: Google Composer.

Предварительные условия для этой новой инфраструктуры следующие:

• Используйте Google Композитор
• Используйте как можно больше инструментов, управляемых Google, для облегчения обслуживания.
• Инфраструктура как код с Terraform
• Отдельные выделенные среды для тестирования
• Весь код, необходимый для выполнения задачи, находится в образе Docker.
• Мониторинг и оповещение в случае сбоя

Таким образом, мы имеем следующую диаграмму:

Схема довольно плотная, разберем ее.

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

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

В нижней части мы находим всю инфраструктуру данных. Существует множество сервисов, управляемых Google.

Мы можем перечислить следующие услуги:

• Секретный менеджер
• Реестр артефактов
• Композитор
• Облачное хранилище
• Облачный IAM
• Облачная регистрация
• Облачный мониторинг
• Большой запрос

А конкретно для среды разработки у нас есть:

Вся установка и настройка инфраструктуры выполняется с помощью Terraform.

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

Кондиционирование рабочей нагрузки

После настройки инфраструктуры с помощью Terraform остается развернуть конвейер ELT, который мы описали ранее. Существует два этапа: извлечение-загрузка и преобразование. Первый шаг выполняет Talend, второй dbt.

Служба Composer использует Kubernetes для запуска Apache Airflow. В нескольких словах, Апач воздушный поток это бесплатное программное обеспечение, которое позволяет выполнять и планировать задачи.

Так что было бы интересно выполнять нашу работу в Kubernetes. Для этого нам понадобится образ Docker.

Talend и dbt упакованы в образы Docker. Вам нужно будет написать файлы Dockerfile и создать образы, которые будут храниться в службе реестра артефактов. Итак, используя оператор KubernetesPodОператор Рабочие нагрузки Talend и dbt, работающие на базе Apache Airflow, выполняются в Kubernetes.

Использование образов Docker значительно облегчает использование различных инструментов, несовместимых со средой Composer.

Особых сложностей, кроме выбора базового образа для Talend, у меня не возникло. Он больше нет изображения официальный OpenJDK JRE. Мне пришлось искать и выбирать образ в одной из организаций, создающих жизнеспособный образ Docker. Базовый образ Docker, предоставленный организацией Adoptium, показался мне наиболее зрелым: https://hub.docker.com/_/eclipse-temurin/

Перейдем к строительству самого трубопровода.

Конвейер с ориентированным ациклическим графом

Talend и dbt — наши два основных строительных блока. Осталось организовать их в файл: DAG. DAG для Направленный ациклический граф или ациклический ориентированный граф на французском языке. Для упрощения граф читается в одном направлении, у него есть начало и конец, и вернуться к началу графа невозможно.

flowchart LR
    talend[Extraction-Chargement Talend] --> dbt_model[dbt model] --> refresh_power_bi[Mise à jour PowerBI]

Эта диаграмма будет преобразована таким образом в Airflow DAG.
from airflow import models
from airflow.providers.cncf.kubernetes.operators.pod import (
    KubernetesPodOperator,
)


with models.DAG(...) as dag:
    talend = KubernetesPodOperator(...)
    dbt_model = KubernetesPodOperator(...)
    refresh_power_bi = KubernetesPodOperator(...)

    talend >> dbt_model >> refresh_power_bi
Находим в DAG оператор KubernetesPodОператори, наконец, порядок задач, которые будет выполнять Airflow.
Создание группы обеспечения доступности баз данных само по себе несложно.  Чтобы понять, как работает Airflow, необходимо понять небольшие тонкости.
Ниже я привожу два из них: разные даты в Airflow и управление ресурсами.
Эти два момента необходимы для понимания того, как работает Airflow.
Дата срабатывания, интервал дат данных
Помимо понятия даты триггера обработки, существуют даты интервалов данных.  Airflow инициирует обработку для диапазона дат данных до даты триггера и в течение следующей даты триггера.
Рассмотрим следующий пример для группы обеспечения доступности баз данных, настроенной с помощью schedule="0 0 * * *".  Воздушный поток должен запускать обработку каждый день в полночь.
На текущий день 18 октября 2023 г. 00:00 UTC

дата срабатывания: «18 октября 2023 г., 00:00 UTC».
дата начала обработки данных: 17 октября 2023 г. 00:00 UTC
дата окончания обработки данных: 17 октября 2023 г., 23:59 UTC
следующая дата срабатывания: «19 октября 2023 г., 00:00 UTC».

Для дополнительной информации, https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html#data-interval
Это понятие не важно в нашем случае использования, но оно имеет значение, когда операции обработки должны извлечь данные за определенный диапазон дат.  Это позволяет вам взять только часть данных, а не все.  Также возможно повторить лечение в течение определенного периода.
Помните, что даты указаны в часовом поясе UTC!  Если ваш кластер Composer запускается в полночь по часовому поясу Европы/Парижа (то есть 22:00 по всемирному координированному времени). накануне), он будет иметь двойную обработку: 1 обработка для интервала дат данных предыдущего дня и 1 другая для интервала дат данных дня запуска Composer.
Управление ресурсами
Управлять ресурсами ЦП и памяти непросто, особенно на неизвестных языках.
Вообще говоря, чем больше ресурсов имеет рабочая нагрузка, тем быстрее она будет обрабатываться.  Так обстоит дело с Талендом.
С 1 процессором и 4 ГБ памяти выполнение было долгим.  Переход на 4 процессора и 8 ГБ сокращает время вдвое.
В DAG это переводится так:
from kubernetes.client import models as k8s_models
from airflow.providers.cncf.kubernetes.operators.pod import (
    KubernetesPodOperator,
)

talend = KubernetesPodOperator(
    (...)
    container_resources=k8s_models.V1ResourceRequirements(
        requests={
            "cpu": "4",
            "memory": "8G",
        },
    ),
)
Диаграммы в Google Monitoring помогли мне внести эти изменения и отслеживать использование ресурсов.
Очень важно иметь систему мониторинга и оповещения как можно раньше в проекте, это позволит вам быстро увидеть эволюцию использования ресурсов и исправить их.
На данный момент эта новая инфраструктура еще не запущена в производство, но у нее есть все необходимые компоненты для ее запуска.
Запущен в производство
Вся эта работа бесполезна, если она не будет запущена в производство.  Чтобы подготовиться к работе, я установил идентичную копию всех исходных наборов данных BigQuery в новую инфраструктуру.  Я воспользовался услугой Передача данных Google.
Затем я написал контрольный список, чтобы ничего не забыть.  Я максимально предугадываю все шаги.  Риск — полная потеря данных при переходе.  Этот список должен быть максимально четким и директивным.  Вы должны уметь раскрываться, не задавая вопросов.
Я синхронизировался с командой, чтобы спланировать запуск.
В важный день список раскрывается.  После завершения производства осуществляется активный мониторинг обработки.  Панель мониторинга проверяется ежедневно.  Как только возникает ошибка, она исправляется в кратчайшие сроки, и снова идет активный мониторинг этого исправления.
А потом что?
После этого выпуска в производство инфраструктура сильно не изменится.  В основном будут проводиться техническое обслуживание и обновления, особенно службы Composer.
Одной из болевых точек на конвейере является Таленд.  Этот инструмент плохо адаптируется к облачной среде.  Проект будет заключаться в поиске альтернативного решения.  Какой инструмент подойдет для извлечения данных и какой будет полностью управляться Google?
В заключение мой отзыв
Создание этой платформы данных было нашим крупным проектом. Студия Одиннадцать Лабораторий.  Все было построено с нуля.  Я хорошо понял проблему, это позволило мне выявить все элементы по функционированию трубопровода.  Решением стала адаптация к его эксплуатации и предпосылкам.  Наконец производство пошло по плану.  Внедрение активного мониторинга позволило мне обнаружить ошибки на ранней стадии.  Это значительно сокращает время простоя платформы.
С моей стороны, эта миссия была очень полной.  я был однажды Архитектор с проектированием инфраструктуры, Операции с написанием Terraform и хорошим пониманием Google Cloud Platform и, наконец, Дев с редакцией DAG Airflow.  Я ухожу с еще большим опытом!