Рубрикатор — о чём почитать на канале

📕#термин_дня — пишу о каком-то термине, специфичном для мира дата аналитики или дата инжиниринга в частности

❔#зачем_нужно — объясняю причину появления "лучших практик" и какие есть риски и дополнительные трудности, если ими пренебрегать

📰 #вести_с_полей — пишу о задачах с работы, интересных и не очень, помогаю составить представление о профессии "изнутри"

🤔 #поразмыслим — интересные технические вопросы, помогающие сформировать "мышление дата инженера"; присылай ответы в комменты

❕#по_запросам — отвечаю постом на вопросы и просьбы подписчиков; присылай свои вопросы в комментарии

👍 #рекомендую — полезные ссылки, сервисы и курсы, которыми пользуюсь сам

📣 #реклама — взаимный пиар с каналами и реклама своих продуктов и услуг

👷‍♂️ #опытным_путём — практика для получения и закрепления навыков

📦 #вперемешку — подборка фактов о технологии или подходе

💭 #отзывы_подписчиков — тёплые слова и пожелания тех, кому канал помог


... другие рубрики грядут ...

#термин_дня

Перекос данных (data skew)

В мире распределённых вычислений данные не помещаются в один сервер, поэтому их разделяют на части и раскладывают по разным нодам (узлам кластера, иногда называют сегментами).

Основные стратегии распределения - по хэш-функции от ключа распределения (не путать с первичным ключом), случайным образом (т.н. round-robin) или полностью копируя датасет на каждую ноду (обычно для таблиц <10 тыс. строк или для небольших файлов, т.н. broadcast).

Перекос случается, если выбрано распределение по хэшу ключа, а сам ключ распределения выбран неоптимально. Если 95% значений = NULL, они попадут на одну ноду. Если 30% сделок совершается с одним VIP-клиентом, они попадут на одну ноду. Если ключ выбран по году совершения продажи и данные есть только по 3 годам, они попадут на 3 ноды, а остальные не будут задействованы.

Поэтому полезно профилировать данные, используя count … group by и общаться с аналитиками для понимания потенциального распределения данных источника в будущем.

#зачем_нужно

Разделение данных на горячие, тёплые и холодные (hot, warm and cold data)

Если спросить пользователей: “С какой задержкой вы хотите получать новые данные”, они всегда рвутся ответить: “С минимальной, в идеале — несколько секунд”. Вопрос о том, готовы ли они за это кратно больше платить, чем за ежедневный батч, остужает пыл.

Схожая история с чтением данных — они бывают разные. Часть требуется для ежедневной работы. Часть запрашивается раз в неделю для совещаний или раз в месяц для регуляторной отчётности. А часть хранится “на всякий случай”, чтобы можно было и через 3 месяца, и через 5 лет обратиться к ним и получить данные с нужной детализацией. Их и называют “горячими” (hot), “тёплыми” (warm) и “холодными” (cold).

Облачные провайдеры зачастую предоставляют 3 или более уровня хранения (tier) на выбор, и чем данные “горячее”, тем быстрее их читать и записывать и тем дороже их хранить. Напротив — чем холоднее, тем дешевле обходится очередной терабайт, но тем дольше и дороже выгрузка данных оттуда. И дата-инженер зачастую сам выбирает уровень, опираясь на нефункциональные требования к репортам и моделям от аналитиков и дата-саентистов. Расходы могут различаться в разы.

Ещё настраивают ротацию данных (rotation, time to live, TTL), когда очередные файлы перемещаются в следующий уровень по превышении отметки в Х штук или в Y дней с момента записи. Также данные могут выносить из OLAP хранилища в S3-бакеты или их аналог.

#вести_с_полей

Ускорил таску пайплайна с 1..7 часов до 0.5-3 минут за счёт уменьшения обращений к базе

Иногда сложно угадать, какой популярностью начнёт пользоваться самописный сервис. И тогда понимаешь, что не просто так на собесах просят оценить асимптотическую сложность алгоритмов в O(n). Даже линейный бывает медленнее, чем нужно.

Сегодня речь пойдёт о сервисе парсинга логов, который унифицирует подключение аналитики студий-партнёров к единому хранилищу данных. Преобразование сырых данных в Parquet’ы и (пере)создание расширяемых вьюшек (представлений, view) в разных разрезах вынесены в Airflow с PyArrow и PySpark. Хранилище - HDFS с Hive’ом, структуры вьюшек хранятся в Redis’е и загружаются в python-объект.

Проблема была в том, что даги Airflow проверяли наличие очередной вьюшки прямолинейной попыткой CREATE OR ALTER. То есть обращений к базе было столько же, сколько вьюшек в принципе, и большую часть времени занимала работа вхолостую. Таких вьюшек за пару лет накопилось полторы тысячи, и даг каждый час пытался ходить полторы тысячи раз в продовую базу и зависать на каждой вьюшке секунд на 15-30.

Что я сделал — нашёл способ обратиться к Metastore Hive’а напрямую из дага, написал SQL-запрос для определения дата-времени последнего изменения DDL-структуры с максимальной фильтрацией только нужных вьюшек через регулярное выражение. Вывел значения в словарь вида {“схема.таблица”: “timestamp”, …}. При каждом реальном изменении структуры вьюшки добавил запись в объект и затем в редис поле с таймстемпом внесения изменений utc_updated_dttm = datetime.now(timezone.utc). Соединил обе части простой условной проверкой: “если есть изменения структуры в объекте и они новее, чем последнее обновление вьюшки, запускай процесс”. Добавил также обработку текущих объектов, где поля utc_updated_dttm ещё нет.

Результат — время холостой проверки, что ничего не изменилось, упало до 30..150 секунд, в зависимости от нагрузки на Hive (один запрос для выгрузки метаданных пачкой и проверка условий в памяти на воркере). До изменений занимало 40..400 минут (n последовательных запросов к Hive, по одному для очередной вьюшки).

С какими проблемами столкнулся:
1. Интеграция задействует много сервисов — HDFS, Hive server, Hive Metastore, Redis, Airflow. Тестового контура нет, в контейнерах поднимать для одной задачи избыточно. Тестировал “грязно” в python-файле с копипастом всех классов и методов, закомментировав все “пишущие” операции. Выводил в STDOUT.
2. Долго дружил таймстемп, прочитанный из Redis’a (строка на основе timezone aware datetime) и таймстемп, прочитанный из Hive (строка из timezone naive datetime), несмотря на UTC в обоих случаях. Пришлось привести строки к datetime объектам через strptime и потом — к epoch time через .timestamp(). Возможно, в будущем перепишу элегантнее через запись в редис в том же формате, в котором читает из хайва.
3. Для отслеживания того, что ложно-отрицательных срабатываний нет (т.е. что изменения пропускаются так же, как их отсутствие), настроил дашборд в Grafana, немного допилив выгрузку в Victoria Metrics. Сами логи для определения конкретной вьюшки с ожидаемыми изменениями смотрел напрямую в докер-контейнере с развёрнутым на проде сервисом.

#термин_дня

Ad-hoc запросы

Спонтанные запросы (зачастую SQL), которые нужны один раз для ответа на один конкретный вопрос. Часть интерактивной и self-service аналитики.

Если вы открываете DBeaver, PgAdmin или другой клиент, пишете запрос, получаете ответ и закрываете, не сохраняя, — это Ad-hoc. Для сохранения обычно создают view и складывают в базу, а не в файлик на рабочем столе. Тогда этим запросом смогут пользоваться коллеги, и он точно не потеряется.

Антонимом ad-hoc являются регулярные (или регламентные) загрузки (запросы, выполняющиеся автоматически по событию или по расписанию).

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

В нереляционном мире это может быть вызов команды “less” или “tail” для файла с логами или чтение значения JSON’а по конкретному ключу из консоли в Redis’e. В общем, любое Preview можно считать ad-hoc запросом.

#термин_дня

Data lineage

Можно перевести как “происхождение данных”, но чаще используется в оригинале. Отслеживание и визуализация того, как данные преобразуются, обычно в виде DAG’а (направленного ацикличного графа). Например, если вью состоит из джоина двух таблиц, то на data lineage можно будет это увидеть. Если таблица наполняется через логический UNION из четырёх разных источников — тоже.

Нужен для ответа бизнес-пользователей на вопрос “как рассчитано это значение” или “из каких источников наполняется эта таблица”.

Бывает на уровне объектов или таблиц (table-level) и на уровне конкретных полей (column-level). Например, в dbt по умолчанию есть первый, а второй подключается как платная опция через DataFold и аналоги.

Data lineage это часть data governance — более-менее научного подхода к организации управления данными в компании. Часто интегрируется вместе с Metadata management инструментами, например с Data catalog.

p.s. произносится как “дейта ли’ниэджь” deɪtə ˈlɪniːɪʤ.

#зачем_нужно

Выбор оптимального количества партиций в Hive

Партиция это директория, в которой лежат файлы. Например, путь до Hive таблицы apps.purchases с двумя партициями app_id и dt может выглядеть так: /raw/apps.db/purchases/app_id=1/dt=2024-01-04/file.parquet. В запросах партиции выступают в роли индексов и резко снижают количество данных для обработки.

Hive тесно связан с HDFS, и тоже оптимизирован под работу с большими файлами. Напомню, что в HDFS данные обычно разделяются на блоки по 64 или 128 МБ. При этом сами блоки хранятся на датанодах, а информация о том, где какой блок лежит и как собрать файл целиком, — на неймноде.

Неймнода немного по-разному работает в разных версиях хадупа, но в общем метаданные о блоках хранятся в памяти. Количество этих записей о блоках ограничено объёмом памяти (расчёт примерно 150 bytes x (1 + количество блоков)). Соответственно, со 128 файлами по 1 МБ кластеру работать сложнее, чем с 1 файлом по 128 МБ.

Для хайв метастора схожая история, каждый файл отслеживается и описывается метаданными. Поэтому если партиции будут слишком мелкими, и в каждой директории будет лежать по файлику, это повысит нагрузку на кластер. С другой стороны — если партиции будут слишком крупными (вся таблица пишется в одну директорию или объединяется по году), запросам нужно будет фильтровать много данных уже после стадии partition pruning’а.

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

Новая рубрика - #поразмыслим

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

Как запустить запрос, который не помещается в ОЗУ?