Airbyte для управления потоками данных –  репликация Яндекс.Метрика в S3

Современные Data Pipelines – это как вода в кране

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

Data Integration / Data Pipelines сегодня стали commodity – они просто должны быть и функционировать, обеспечивая базовые потребности, при этом основной фокус работы Аналитиков и Инженеров приходится на моделирование данных, трансформацию, обогащение, агрегирование, а также визуальную подачу выводов.

Второй важной особенностью является смена парадигмы от ETL к EL(T). Я попробую изложить ключевые идеи в паре тезисов.

Исторический подход ETL предполагал последовательность Extract – Transform – Load, что выявило ряд проблем:

• Нет прозрачности – в Хранилище попадают уже трансформированные данные, без возможности восстановить историю и исходные данные

• Отсутствие гибкости – трансформации должны быть известны и разработаны заранее, любые изменения и дополнительные требования могли стоить дорого

• Зависимость Аналитиков от Инженеров (иногда очень скилованных) – сложные варианты интеграции с рядом источников, высокий порог сложности решений 

Современный подход EL(T) предполагает независимые этапы Extract – Load и Transform:

• Гибкость – из сырых исходных данных можно собрать что угодно, какие бы идеи у вас не возникали, даже если они часто меняются

• Вычислительные ресурсы и хранение данных доступны как никогда – Облачные сервисы хранить все данные без необходимости экономить

• Разделение этапов EL и T – вы больше не завязаны на один инструмент, но вправе использовать любые тулзы для трансформации данных, такие как dbt, Airflow.

Множество компаний сегодня предоставляют Data Pipelines / Integration как сервис. Перечислю те, с которыми мне доводилось сталкиваться: Fivetran, Hevo, Alooma, Stitch.

Их основные преимущества:

• Надежность и поддержка от вендора

• Полностью управляемый сервис – минимум забот на вашей стороне

• Легкая конфигурация pipelines – все стремятся упростить настройку

Но есть и ряд недостатков:

• Это закрытый код – вы ограничены возможностями которые поддерживает вендор

• Могут найтись специфические коннекторы (или способы подключения), которые вендор не поддерживает

• И конечно это стоимость – чек может быть очень большим

Альтернативно, существуют класс современных и удобных решений для управления потоками интеграции данных с открытым исходным кодом: Airbyte, Meltano, Singer. И вот одно из таких решений сегодня я и предлагаю рассмотреть.

И да, честь и хвала разработчикам и контрибьюторам таких решений.

Airbyte – простота и гибкость в интеграции данных

Airbyte – это проект с открытым исходным кодом, который стремительно набирает популярность. Проект доступен на ​​GitHub (3.800+ stars), а сообщество в Slack насчитывает 2.500+ человек. По сути это современный стандарт для выстраивания потоков интеграции данных из всевозможных приложений, баз данных и API в аналитические хранилища данных, озера данных. Ниже я коротко рассмотрю ключевые преимущества инструмента.

Обширный набор коннекторов, доступных для подключения в считанные минуты. В списке все самые популярные СУБД, а также огромное количество популярных сегодня приложений: Intercom, Zendesk, Stripe, Salesforce, Jira. Усилиями сообщества пользователей список коннекторов постоянно растет. Добавление новых коннекторов сведено к простому конфигурированию – оркестрацией и спосбоами репликации займется Airbyte.

Понятная и масштабируемая архитектура. Хранилище метаданных, в качестве которого можно использовать внешнюю СУБД (Postgres), веб-интерфейс, набор рабочих лошадок (Workers), число которых можно гибко регулировать, а также полноценный scheduler с возможностью гибко регулировать частоту репликации данных.

Различные варианты установки приложения: AWS, Azure, GCP, K8s, Docker. Подходящий вариант для компаний, которым необходимо разместить приложение на своих мощностях в связи с требованиями к безопасности и compliance. При размещении в облаке – данные хранятся в вашем облаке и стоимость ресурсов остается прозрачной.

Различные стратегии синхронизации данных – Sync strategies:

• Full Refresh Overwrite: полная выгрузка всего объема данных и перезапись на приемнике

• Full Refresh Append: полная выгрузка всего объема данных и добавление на приемнике

• Incremental Append: инкрементальное чтение записей и добавление на приемнике

• Incremental Deduped History: инкрементальное чтение записей, добавление на приемнике, а также формирование дедуплицированной версии представления

• Manual full refresh: в случае необходимости провести полную репликацию данных из источника

Нормализация данных и преобразование типов. Также Airbyte может быть полезен в переводе массивов и вложенных (nested) коллекций в плоские структуры.

И, пожалуй, одно из самых главных – вы платите только за используемые вычислительные мощности (в случае использования облака). Никакой платы за количество коннекторов и объем реплицируемых строк, как в сервисах типа Hevo, Fivetran.

Развертывание Airbyte в Yandex.Cloud

Предлагаю попробовать пощупать Airbyte своими руками, развернув приложение в Яндекс.Облаке. Для этого нам понадобится проделать ряд шагов:

• Регистрация в Облаке

• Создание виртуальной машины

• Создание пары ключей и подключение по SSH

• Установка Docker + Docker compose

• Запуск Airbyte

• Конфигурирование Yandex Object Storage (S3) – опционально

# Начало работы с интерфейсом командной строки # Install: https://cloud.yandex.ru/docs/cli/quickstart#install # Init profile: https://cloud.yandex.ru/docs/cli/quickstart#initialize   yc config list # check   # Подключиться к виртуальной машине Linux по SSH # Создание пары ключей SSH: https://cloud.yandex.ru/docs/compute/operations/vm-connect/ssh#creating-ssh-keys   ls ~/.ssh/id_rsa.pub # check   # Создать виртуальную машину из публичного образа Linux # https://cloud.yandex.ru/docs/compute/operations/vm-create/create-linux-vm   # проверьте путь до файла с публичным ключом yc compute instance create \    --name airbyte-vm \    --ssh-key ~/.ssh/key.pub \    --create-boot-disk image-folder-id=standard-images,image-family=ubuntu-1804-lts,size=10,auto-delete=true \    --network-interface subnet-name=default-ru-central1-a,nat-ip-version=ipv4 \    --memory 2G \    --cores 2 \    --hostname airbyte-vm     yc compute instance list # check # Запишите публичный адрес ВМ – EXTERNAL IP   # Подключение к виртуальной машине # https://cloud.yandex.ru/docs/compute/operations/vm-connect/ssh#vm-connect   ssh -L 8000:localhost:8000 -i ~/.ssh/key yc-user@<external-ip> # укажите свой EXTERNAL IP   # Install docker   sudo apt-get update sudo apt-get install -y apt-transport-https ca-certificates curl gnupg2 software-properties-common curl -fsSL https://download.docker.com/linux/debian/gpg | sudo apt-key add -- sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/debian buster stable" sudo apt-get update sudo apt-get install -y docker-ce docker-ce-cli containerd.io sudo usermod -a -G docker $USER   # Install docker-compose   sudo apt-get -y install wget sudo wget https://github.com/docker/compose/releases/download/1.26.2/docker-compose-$(uname -s)-$(uname -m) -O /usr/local/bin/docker-compose sudo chmod +x /usr/local/bin/docker-compose docker-compose --version   # Install Airbyte   git clone https://github.com/airbytehq/airbyte.git cd airbyte git checkout v0.29.2-alpha sudo docker-compose up -d   # Access at http://localhost:8000/preferences 

В случае необходимости, Airbyte можно масштабировать (Scaling) следующими способами:

• Увеличение количества Workers – scaling out

• Использование более мощного типа VM (CPU, Memory, Disk) – scaling up

Подключим источник данных – Yandex.Metrika

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

Интерфейс позволяет выбрать интересующие метрики, разрезы и сегменты (фильтры) и представить данные в наглядном виде в интерактивном режиме (под капотом Clickhouse, который не тормозит!). Ознакомиться с ним можно по ссылке Metrika Live Demo.

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

Итак, в базовом виде обращение к API конфигурируется следующим образом:

• Тип выгружаемого отчета – таблица, drill-down, time-series

• Набор интересующих измерений, метрик, сегментов

• Формат ответа и семплирование

• Параметризация (валюта, атрибуция, цели)

В качестве обращения предлагаю формировать отчет Источники, сводка: https://api-metrika.yandex.net/stat/v1/data?preset=sources_summary&id=44147844

Для этого в Airbyte необходимо создать новый источник (source):

• Тип  – HTTP Request

• Url – API Endpoint

• Метод – GET

Конфигурируем простой pipeline и устанавливаем расписание

В качестве приемника данных будем использовать объектное хранилище Yandex Object Storage (совместимое с S3).

Для конфигурации бакета проделаем следующие шаги:

## Configure S3 Access # Creating a service account: https://cloud.yandex.com/en-ru/docs/iam/operations/sa/create # Assign a role to service account: https://cloud.yandex.ru/docs/iam/operations/sa/assign-role-for-sa # Create access key: https://cloud.yandex.com/en-ru/docs/iam/operations/sa/create-access-key # Access with AWS S3 CLI: https://cloud.yandex.ru/docs/storage/tools/aws-cli   yc iam service-account create --name airbyte-s3 yc iam service-account list yc iam service-account get airbyte-s3   yc iam service-account add-access-binding airbyte-s3 \    --role storage.admin \    --subject serviceAccount:<id>   yc iam access-key create --service-account-name airbyte-s3   aws configure --profile yc   aws --profile=yc --endpoint-url=https://storage.yandexcloud.net s3 mb s3://<bucket-name> aws --profile=yc --endpoint-url=https://storage.yandexcloud.net s3 ls 

Настроим Destination в Airbyte:

• Destination type – S3

• Endpoint – https://storage.yandexcloud.net

• Имя Bucket и путь

• S3 Key Id & Access Key полученные при настройке технической учетной записи

Обратите внимание на то, что помимо JSON из коробки доступен ряд других файловых форматов:

• CSV

• AVRO (binary) – гибкий, для schema evolution

• Parquet – бинарный, колоночный, оптимизированный под чтение

Для финализации пайплайна выберем расписание и тип репликации:

• Раз в сутки

• Full refresh – Append

После сохраним и запустим репликацию.

Изучим выгруженные данные

Выгрузка представляет собой JSON-документ с:

• параметрами обращения

• результатами запроса

• метаданными о выгрузке

{    "query": {        "ids": [            44147844        ],        "preset": "sources_summary",        "dimensions": [            "ym:s:lastSignTrafficSource",            "ym:s:lastSignSourceEngine"        ],        "metrics": [            "ym:s:visits",            "ym:s:users",            "ym:s:bounceRate",            "ym:s:pageDepth",            "ym:s:avgVisitDurationSeconds"        ],        "sort": [            "-ym:s:visits"        ],        "date1": "2021-08-19",        "date2": "2021-08-25",        "limit": 100,        "offset": 1,        "group": "Week",        "auto_group_size": "1",        "attr_name": "",        "quantile": "50",        "offline_window": "21",        "attribution": "LastSign",        "currency": "RUB",        "adfox_event_id": "0"    } } 

{    "total_rows": 116,    "total_rows_rounded": false,    "sampled": false,    "contains_sensitive_data": false,    "sample_share": 1.0,    "sample_size": 2698,    "sample_space": 2698,    "data_lag": 184,    "totals": [        2694.0,        2363.0,        36.82256867,        1.63437268,        85.72345954    ],    "min": [        1.0,        1.0,        0.0,        1.0,        0.0    ],    "max": [        833.0,        736.0,        100.0,        9.0,        1201.0    ] } 

{    "data": [        {            "dimensions": [                {                    "icon_id": "0",                    "icon_type": "traffic-source",                    "name": "Direct traffic",                    "id": "direct"                },                {                    "name": null,                    "id": null,                    "favicon": null,                    "url": null                }            ],            "metrics": [                833.0,                736.0,                28.81152461,                1.7334934,                105.03601441            ]        },    ...    ] } 

Ради интереса можно попытаться найти соответствие чисел в веб-интерфейсе Я.Метрики и выгрузке через API.

Понравилось – хочу еще

Браво! Поздравляю с первым потоком интеграции на реальных данных и использованием Modern Data Stack.

Еще больше современных инструментов мы изучаем на занятиях курсов Data Engineer и Analytics Engineer в OTUS: dbt, Clickhouse, Dataproc, Airflow. Но главное – то, что я и мои коллеги стремимся дать полноценную картинку и практические навыки, так необходимые для работы.

Главные преимущества такого подхода:

• Живое общение на регулярных вебинарах

• Пошаговые практические инструкции и домашние задания

• Обратная связь и возможность получить консультации к своим решениям

Data Engineer – один из самых успешных тиражных курсов, в запусках которого я участвую уже более 2-х лет. К новому старту готовы кардинальные обновления по содержанию, используемым инструментам, инфраструктуре, включая выделенные вебинары на разбор домашних заданий.

Analytics Engineer – попытка закрыть потребность на людей-мультиинструменталистов, которые сильны и в понимании специфики бизнеса, моделировании и в инженерной части. Львиная доля курса посвящена современным аналитическим СУБД, BI-инструментам, практикам продвинутой аналитики и моделирования в dbt.

Спасибо за внимание!