Теория или практика?

📖 Наконец-то прочитала книжку "Spark в действии". Очень понравился вот этот рисунок: он супер понятно описывает суть ленивых вычислений) Spark копит все преобразования над данными, пока что-то не заставит его запустить весь пайплайн (например, show).

Отметила для себя пару новых моментов:

1️⃣ Можно создать табличку через sql и сразу перевести её в датафрейм (я думала, что такое работает только для реальных таблиц)

Например:

spark.sql(
  'select id from users'
) \
.createOrReplaceTempView('view_users')
df_users = spark.table('view_users')

2️⃣ Есть интересные методы сортировки asc_nulls_first, asc_nulls_last (для desc так же), чтобы обозначить порядок для нуллов

🐰 Ещё позабавили некоторые формулировки)

"Агрегации - ... в известной степени приближение к методам машинного обучения и искусственного интеллекта".

"Во время моей работы со Spark я встречал два типа людей: тех, кто имел опыт работы с Apache Hadoop, и тех, у кого не было такого опыта".

Надеюсь, что это трудности перевода))

🐱 Вообще пришло осознание, что теоретические книжки легко читаются, когда ты уже пощупал это на практике.

А вы как считаете, когда нужно читать умные книжки?

#spark

Репликация табличек

👫Кратко про стек
clickhouse - это база данных))
airflow - оркестратор)
dbt - тулза для удобной работы с трансформацией данных, контроля их качества, ещё позволяет все DDL делать через селекты

💐Кейс
Недавно у нас данные в кликхаусе переехали на новый кластер с поддержкой репликации. А поверх этих данных крутились пайплайны в airflow и запускались проверочки на dbt. И в один момент все упало😐

Что ж, надо фиксить. Самая большая боль - это как поднять упавшие dbt модельки, которые писали данные в таблички. Например, есть такие, которые каждый день собирают количество строк в данных для мониторинга.

Пошел пул проблем:

🔴откуда берутся таблички с суффиксами dbt_backup, dbt_tmp_tables, dbt_new_data и почему они затирают оригинальные

🔴почему инкрементальные модельки (которые дописывают порцию данных) не могут это делать при повторных запусках

🔴как создавать таблички из csv файликов с движками Replicated

🔴почему вдруг я не могу запускать больше 5 параллельных тасок в даге и как это исправить

🔴нет грантов на операцию sync replica

🔴почему при чтении одной и той же таблицы то 100 строк, то 0
...

Задала вопрос в сообществе dbt в слаке, и там один чел прокомментил, что столкнулся с теми же самыми проблемами. Совпадение? (он из WB)

В общем, это всё удалось решить так:

⭐️в dbt обязательно прописывать в конфигах cluster (спец. штука для реплицированных таблиц)

⭐️для csv файликов тоже в конфигах можно прописать дефолтный движок

⭐️для параллельного запуска тасок есть параметр max_active_tasks

⭐️перепроверить, что в коде с модельками и сами таблички заданы через Replicated

Многие проблемы решились последним пунктом, в том числе с 0 строк. Тут суть в том, что когда создается стандартная табличка (с движком без Replicated), то она находится на одной ноде. Но когда мы запросом туда стучимся, то нас могут отправить на другую ноду, где этих данных нет👍

А вы работали с репликацией, есть запомнившаяся мудрость на будущее?

P.S. все, идите отдыхайте 😶🐱🐱

🔍Задачка на поиск

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

💬Описание
На фронте есть табличка с именами и датами, по которой есть поиск. Поиск нужен по столбцу дат. Поисковой запрос идёт на бэк, и грузятся данные из бд.

💫Особенности (как же без них)
Дата в формате mm/dd/yyyy: 05/04/2024 = 04.05.2024. В бд как обычно: 2024-05-04.

Примеры
1️⃣ Ввод: 11/20
Вывод:
🔘mm/11/20 - любой месяц
🔘11/20/yyyy - любой год

2️⃣ Ввод: 1/2
Вывод:
🔘mm/_1/2___ - дни могут быть 01, 11, 21, 31 (если есть), год - начинается на 2
🔘_1/2_/yyyy - месяц 01 или 11, день - с 20 по 29

💐Что нужно?
Предложить наиболее эффективный вариант поиска.

Какие идеи?

💡Ответ💡

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

❔Что это?
Это такой индекс, в котором ключи посчитаны через функцию.

😎Решение №1
Никогда его не использовала, но для нашего случая будет по типу:

CREATE INDEX us_date_idx ON my_table (concat_ws('/', month(dt), day(dt), year(dt)))

Делаем 'concat like входная строка' - применяется наш индекс.

👉Решение №2
Если мы не можем создать индекс, идея такая:

1. Разделить поисковую строку по /
2. Три случая:

▪️1 элемент - день, месяц или год

like '%elem1%'

▪️2 элемента - либо день и год, либо месяц и день

--день и год 
--1/2 - это конец дня и начало года 
like 'year%day' 
 
--месяц и день 
--1/2 - это конец месяца и начало дня 
like '-%month-day%' 

▪️3 элемента

--1/29/202 - месяц и год могут быть недописаны 
like 'year%-%month-day'

💕dbt deep dive💕

У нас в команде есть история с deep dive. Мы выбираем тему, собираем все наши знания и задачи, раскапываем всё. Потом делаем презу и просвещаем коллег.

Эта преза моего дип дайва по dbt. Там рассказывается:
✔️что делает этот инструмент
✔️как создать проект
✔️что в этом проекте есть
✔️как запускать

А вы какие фишки используете?

Как заруинить джойн в спарке?

С помощью udf!

🧐 Что это?
UDF (user defined function) - это функция, которую мы сами написали и можем применить в спарке (и не только).

🤩 Пример
Самый обычный джойн:

df1.join(df2, ['id'], 'inner')

План запроса - стандартный SortMerge (без деталей):

== Physical Plan ==
+- Project
   +- SortMergeJoin
      :- Sort
      :  +- Exchange hashpartitioning
      :     +- Filter isnotnull
      :        +- Scan ExistingRDD
      +- Sort ...

😭 Перепишем запрос через udf:

compare_udf = F.udf(lambda x, y: x == y, BooleanType()) 

df1.alias('df1') \
.join(
    df2.alias('df2'),
    compare_udf(
        F.col('df1.id'),
        F.col('df2.id')
    ),
    'inner'
)

И все - теперь у нас под капотом декартово произведение:

== Physical Plan ==
*(3) Project
+- *(3) Filter pythonUDF0#56: boolean
   +- BatchEvalPython [<lambda>(id#0L, id#4L)], [pythonUDF0#56]
      +- CartesianProduct
         :- *(1) Scan ExistingRDD
         +- *(2) Scan ExistingRDD

👀 А все потому, что для спарка udf - это черный ящик, и он не будет заглядывать вовнутрь. Так что схема такая:

SortMerge -> Cartesian
ShuffledHash -> Cartesian
BroadcastHash -> BroadcastNestedLoop

#spark

🧳 Рабочий кейс

Недавно с коллегой обсуждали задачу:

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

1️⃣ Первая версия кода была такая - каждой функции по циклу:

for dt in pd.date_range(start_dt1, end_dt):
  load1(start_dt1)

for dt in pd.date_range(start_dt2, end_dt):
  load2(start_dt2)

И таких функций 10. 20. А Хотелось компактно организовать🎁

2️⃣ Вторая версия - итерируемся от минимальной даты и навешиваем условие перед запуском каждой функции:

min_dt = min(start_dt1, start_dt2)

for dt in pd.date_range(min_dt, end_dt):
  if dt >= start_dt1:
    load1(dt)
  elif dt >= start_dt2:
    load2(dt)

Тоже с условиями слишком громоздко🏔

3️⃣ В итоге мы дошли до такой финальной версии:
⭐️создаем функцию для генерации цикла по датам
⭐️создаем список функций с датами
⭐️передаем функции как аргументы

def loop(start_dt, end_dt, func):
  for dt in pd.date_range(start_dt, end_dt):
    func(dt)


func_list = [(load1, start_dt1), (load2, start_dt2)]

for func, dt in func_list:
  loop(dt, end_dt, func)

Чтобы добавить новую функцию, теперь достаточно дописать ее в func_list.
А у вас есть идеи элегантного решения?

🔠🔠🔠Unit-тесты для данных
появились в версии dbt 1.8

https://docs.getdbt.com/docs/build/unit-tests

https://github.com/dbt-labs/dbt-core/releases/tag/v1.8.0

Уже успели попробовать?

Обработка больших данных

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

Если считать в лоб, то надо пройтись по ~ 1 млрд * 6 месяцев * 30 дней = 180 млрд строк 🔥

Вы запустили запрос, все зависло, кластер упал, оборудование перегрелось, дата центр сгорел - вас уволили и заставили влезть в кредиты😱😱😬😬 ну, или нет)

❔ Что делать?

Такие таблицы обычно считаются по дням. Посчитали день - положили в датафрейм. Потом новый кусок объединили с предыдущими.
Или так: посчитали - сохранили в таблицу. И каждый новый день по очереди дописывается.

❕ Поэтому семь раз режем, один раз мерим.✂️

Пул мудростей

Собрали с коллегами по несчастью папочку с полезными каналами про данные

Велком, если хотите увеличить площадь вашего погружения) 🐤

Логи расчетов

Просто оставлю здесь это искусство🎨

😊 Розыгрыш! 😊

Разыгрываю книжку "Spark в действии", которую даже сама читала в электронном варианте😁
Готова переслать по России или лично вручить в Мск!

Нужно:
1️⃣ Быть подписанным на канал
2️⃣ Оставить коммент - напишите любое ключевое слово из sql/spark/языков программирование/airflow итд итп

Итоги будут подведены 17.06.2024

Удачи🙂

Как ПСИ проходили

🦆 Мы сейчас мигрируем витрины данных на свой кластер и перед каждым релизом проходим ПСИ.

🤓 ПСИ - это приемо-сдаточное испытание, где проверяют, а что за данные, а как, а почему, а с какой целью, а какой атрибутивный состав, а какая логика расчета и тому подобное. Что потоки с витринами успешно запускаются без падений.

👉 И вот во время созвона мы обнаруживаем, что для установки на прод не хватает release notes, которые автоматом проставляются. Т.к. это все для нас впервые, то мы не в курсах, как их быстро добавить.

🏃‍♂️ Пишу коллеге, который нам помогал по этому поводу, - не читает. Звоню(!):

- Как проставить release notes?
- Надо запустить доп. пайплайн в Jenkins
- Где его найти?
- Лежит рядом с основным (хотя бы это знаем)

Иду в Jenkins, как заполнять поля - непонятно, смотрю примеры, делаю по аналогии, запускаю - отработало!!!

👨‍🦯 Лид тайм от реагирования до исполнения: 16 минут.
ПСИ успешно пройдено.

Mic Drop

Что спросить на собесе?
У HR, команды и руководителя.

Недавно общались с коллегами и накидали свой список, что бы мы спросили/что обычно спрашивают:

1️⃣ Офис/удаленка/гибрид
Условия согласования
Обязательно ли с 9 до 18 или можно двигать, дорабатывать попозже

2️⃣ Выдается ли техника?
Как вам доставят?
Есть ли офис в вашем городе?

3️⃣ Новая команда или на замену?
Команда может укомплектовываться, но может и разваливаться

4️⃣ Состав команды
Сколько человек?
Сколько дата инженеров/аналитиков, других ролей?
Разбивка по джунам/мидлам/сеньорам?

5️⃣ Кто обучает?
Кто лидит в команде: техлид, тимлид, продукт оунер?
Кто делает code review?

6️⃣ Аутсорс/аутстаф?

7️⃣ Какие инструменты, технологии?
Скорее всего, к этому моменту вам уже все рассказали

8️⃣ Скрам или канбан?
Канбан означает, что команда типо сервисная, много ad hoc-ов
Есть ли Jira, Confluence или что-то подобное?

9️⃣ Есть ли системный аналитик?
Если нет, то придется вам копаться в документации и общаться с заказчиками

1️⃣0️⃣ Кто заказчики?
От кого ТЗ?
Аналитики, бизнес, другие подразделения

1️⃣1️⃣ Описание рабочего дня
Какие задачи?
Бывают ли переработки?

1️⃣2️⃣ Какие KPI для сотрудника?
Как расти в компании?

1️⃣3️⃣ Есть ли скрам-мастер?
Проводится ли ретро?
Глобальное демо на несколько команд? (например, раз в квартал)

1️⃣4️⃣ Культура командных посиделок
Есть ли чатик с мемами?^^

🐾Есть еще что добавить?

🍄 Итоги розыгрыша 🍄

Кое-кому сегодня удача особенно подмигнула,
и нашим победителем становится...

Улмас Киличов ☺️☺️

Напишу, обсудим все детали.
И буду думать, что бы еще интересного замутить)

А пока всем хорошего дня и недели!

🔵Колоночные базы vs Семейство колонок

У нас закупаются корпоративные курсы, и я сейчас прохожу очередной по де) До этого не сталкивалась с семейством колонок, везде пишут по-разному, так что держите ликбез 🐹

❔ Колоночная/столбцовая СУБД
Columnar, Column-oriented DB

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

Примеры: ClickHouse, Vertica, Snowflake, Google BigQuery.

❔ Семейство колонок
Wide-column store

А вот это уже NoSQL штука.

🖌Пример - таблица с пользователями.

У каждой строки есть ключ:

Maria

Для каждого ключа задается семейство колонок:

personal_info, address

Для каждого семейства - сами колонки:

personal_info <- firstname, lastname, age
address <- city, street

Сами колонки можно добавлять на лету, заполнять у строк по-разному или вообще оставлять пустыми (null не хранится).

Примеры: HBase, Cassandra, Google BigTable, ScyllaDB.

Тут статейка очень старенькая, но есть полезности.

😐 Я поняла семейство колонок как подтаблицу. Они не поддерживают джойны/подзапросы, поэтому OneToMany и все остальные моделируются вот так.

По терминологии так до конца и непонятно. Вроде как и те, и другие - колоночные. Но первые реляционные, а вторые - нет. Как-то так)

Вопросы на middle+

😵 Недавно позвали на собес, решила сходить для контента🤩 Для меня он был прям сильно нестандартным, т.к. больше гоняли по кейсам "что будешь делать, если" и задавали нетривиальные вопросы по технике.

🚩 Ниже краткий списочек тем и вопросов:

DataOps
Синхронная/асинхронная репликация
Селективный запрос
Покрывающий индекс
Всегда ли план запроса отражает правду?
Zookeeper
MapReduce
Spark
Что делать, если надо показать отчет за месяц, а данных за один день нет?
Что делать, если после загрузки в dwh данных в 3 раза больше?
Золотая запись
PL/SQL процедуры
Движки в КХ
MDM (master data management)
Data Lineage
Data Vault/Anchor/Звезда
В каких случаях что выбирать?
ACID
WAL (Write-Ahead Logging)
Что сложнее - update vs insert?
Физические джойны
Какую бд выберешь, если ...?
Что делать, чтобы сравнить данные с источником, api?
Как будешь решать нестандартную задачу?

👍 Можете порассуждать в комментах)

#собес

Вышел подкаст со мной!
Теперь я звездочка 😊