дата инженеретта

Всем привет!

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

Сколько у вас опыта работы ✨в сфере данных✨?

💡Сортировка в Spark💡

Правильные ответы - все, кроме 'id'.desc()

sort vs orderBy - что вам больше нравится, никакой разницы

Способов тоже много на ваш выбор:

1️⃣ F.col().desc()

Через метод столбца. Мы можем навесить функцию только на объект типа колонка, на обычную строку - нет:

df.orderBy(F.col('id').desc())

2️⃣ ascending=False

Через свойство, в сортировке - строка или колонка:

df.orderBy('id', ascending=False)
df.orderBy(F.col('id'), ascending=False)

3️⃣ F.desc()

Через функцию, аргумент - строка или колонка:

df.orderBy(F.desc('id'))
df.orderBy(F.desc(F.col('id')))

4️⃣ spark.sql

Для особо ленивых)

spark.sql('select * from my_table order by id desc')

#spark

🤭Вредные советы

Недавно ходила на сходку ODS (Open Data Science) и познакомилась с одним дата аналитиком. В общих чертах состоялся такой диалог:

- Чем вы занимаетесь?
- Шатаю таблички.
- И каким образом?
- Да просто select *, и все)

Так вот этот пост про do и don'ts в сфере работы с данными:

1️⃣ Всегда делайте SELECT *, а не только нужные поля - вдруг они пригодятся в будущем? И никаких LIMIT - мы не хотим делать выводы на крошечной выборке

2️⃣ Никогда не навешивайте индексы, они просто забивают память

3️⃣ Добавляйте как можно больше джойнов в один запрос - надо экономить место

4️⃣ Никогда не проверяйте, какие типы данных сопоставляются в ON, WHERE и т.д. - лучше сделайте побыстрее и идите отдыхать

5️⃣ Считайте агрегации несколько раз вместо использования cte или temp табличек - это чисто синтаксический сахар

6️⃣ Cross join - наше все

7️⃣ Вставляйте как можно больше OR, не пытайтесь заменить на IN, UNION и т.д.

8️⃣ Если нужен DISTINCT, он должен быть в каждом подзапросе - для нашей 200% уверенности

9️⃣ Навешивайте на фильтруемые поля кучу функций - UPPER, LOWER, LEFT, RIGHT... Ну а

WHERE UPPER(name) LIKE '_Mary%'

- вообще песня!
_ ⏩1 символ
%⏩0 или много символов

🔟 Чем больше подзапросов - тем выше ваша зарплата💵

💡Ответ💡

Правильные варианты - со 2 по 7.

При конвертации UTC -> UTC+3 добавляется 3ч ко всем строкам, поэтому период начинается с 21:00 (UTC) = 00:00 (UTC+3) и заканчивается тоже в 21:00, потому что в between самое крайнее время - это не конец дня, а начало дня:

--это одинаковые условия
BETWEEN '2024-02-24' AND '2024-02-25'

BETWEEN '2024-02-24 00:00:00' AND '2024-02-25 00:00:00'

Зачем?
🍊Как-то раз мне нужно было проверить, долетают ли данные из одной системы в другую. Но каунты тотально не сходились.

Потом я увидела, что день начинался в 21:00 предыдущего дня и заканчивался в 20:59 текущего. Так, разница в 3 часа, значит, связано с конвертацией дат в UTC - но это было не совсем так!

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

Я работала в DBeaver, а когда написала тот же запрос в PyCharm - даты отображались корректно, с 00:00 до 23:59.

Возможно, во всем виноваты настройки, но так и не получилось с ними разобраться. PyCharm люблю всей душой👩‍💻

Код потестить (ClickHouse):

CREATE TABLE dates (
  `datetime` datetime('UTC')
)
ENGINE = MergeTree()
ORDER BY datetime;

INSERT INTO dates VALUES
('2024-02-23 20:59:00'), ('2024-02-23 21:00:00'), ('2024-02-23 23:59:00'), ('2024-02-24 00:00:00'), ('2024-02-24 02:59:00'), ('2024-02-24 20:59:00'), ('2024-02-24 21:00:00'), ('2024-02-24 22:00:00'), ('2024-02-25 00:00:00'), ('2024-02-25 02:59:00');

SELECT
  toDateTime(`datetime`, 'Europe/Moscow'),
  CASE
    WHEN toDateTime(`datetime`, 'Europe/Moscow') BETWEEN '2024-02-24' AND '2024-02-25' THEN 1
    ELSE 0
  END AS flag
FROM dates
ORDER BY 1;

Про личный бренд

😯Недавно у нас стартанул курс по амбассадорству "Академия амбассадоров Сбера", где я буду учиться на протяжении 6 недель, и это моя домашка)

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

Планирую пробовать новые форматы (даже если некомфортно) и прокачивать себя. Как говорится, "Иди туда, где страшно"🌝

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

▶️Так что когда-нибудь обязательно ждите на экранах ваших смартфонов!
Пока все!

💡Ответ 💡

👨‍💻Немного поресерчила, погоняла запросы в песочнице в постгре и на кластере на спарке. Оказалось, что не все однозначно (как и показали результаты нашего опроса). Это зависит от:
🔘базы и работы оптимизатора
🔘количества данных
🔘логики данных (может, юнион нам вообще не подходит)

Смотрим план!

1️⃣: Nested Loop/Cartesian Product
2️⃣: Hash Join/SortMergeJoin

Кажется, что union сильно лучше, т.к. не происходит комбинаторного взрыва. Когда я работала на MS SQL, то сталкивалась с подобными запросами с OR в условии джойна, и они подвисали конкретно. Но на медиуме есть статейки/комменты для обеих сторон.

В общем, it depends)
Смотрите планы запросов, замеряйте время, понимайте логику данных, выбирайте то, что лучше при ваших условиях😌

Выходные прошли, погнали снова работать)

Есть ли разница в запросах?

--1
SELECT *
FROM table1 t1
  JOIN table2 t2
    ON t1.id = t2.id or t1.name = t2.name;

--2
SELECT *
FROM table1 t1
  JOIN table2 t2
    ON t1.id = t2.id
UNION ALL
SELECT *
FROM table1 t1
  JOIN table2 t2
    ON t1.name = t2.name;

Опрос ниже👇

Работа с таблицами в Hive

В Hive есть два типа таблиц:
🔘internal (managed)
🔘external (not managed)

Managed означает, что Hive в ответе за тех, кого приручил🌹

😑Internal таблица похожа на обычную таблицу из любой СУБД - она хранит данные и метаданные (в Hive Metastore) о них. Туда можно вставлять данные, из неё можно читать. А ещё она поддерживает кэширование (переиспользование уже посчитанного запроса в других) и, начиная с определённой версии движка, ACID-транзакции. Если удалить такую таблицу, данные тоже удалятся, тут как обычно.

Пример создания таблицы, партиционированной по дате, которая хранится в виде паркет-файлов:

CREATE TABLE web.visits (
  visitid int,
  url string
)
PARTITIONED BY (dt string)
STORED AS PARQUET;

🤗А еще есть External таблица, которая хранит только метаданные о том, где лежат файлики и как их оттуда читать. Многие функции вроде ACID-транзакций и кэширования не поддерживаются. Если таблицу удалить, данные останутся лежать на дисках.

По созданию то же самое, нужно только дописать ключевое слово external:

CREATE EXTERNAL TABLE web.visits (...)

Кстати, на днях мы пытались считать данные из таблицы через оркестратор и столкнулись с ошибкой "Your client does not appear to support insert-only tables". Оказалось, что по дефолту у нас на кластере таблицы создаются транзакционными и нужно добавить опцию:

ALTER TABLE web.visits SET TBLPROPERTIES('transactional'='false');

😶При этом эти настройки не сработали с managed табличкой, вероятно, при настройке кластера добавили параметр hive.strict.managed.tables=true, который запрещает создание управляемых и нетранзакционных таблиц. Но с external этот метод сработал!

👫Если вы хотите узнать побольше о том, какие полезные команды используются в Hive почти каждый день, смотрите вторую часть поста https://t.me/rzv_de/132
А еще можете пройтись и по другим заметкам)

#hive

🧩Работаем с партициями

➗Вообще партицирование/партиционирование - это метод разделения большой таблицы на маленькие кусочки для оптимизации обращения к ним. Часто партицируют по дате или по полю с небольшим конечным множеством значений. В HDFS они выглядят как подпапочки внутри одной папки:

hdfs://data/web/visits/ - вся таблица
hdfs://data/web/visits/visit_date=2024-03-01 - конкретная партиция
А уже внутри этого пути лежат паркет-файлики вида part-23001-b484-8348bde71.c000.parquet

Как эффективно достать данные из партицированной таблицы?

У нас есть табличка visits в схеме web, и она партицирована по дате. Нужно вытащить максимальную дату.

Самый простой способ - написать запрос:

spark.table("web.visits") 
 .select(max("visit_date")) 
 .show() 

Но мы также можем воспользоваться преимуществами Hive и сделать так:

spark.sql("SHOW PARTITIONS web.visits") 
 .select(max("partition")) 
 .show()

В результате show partitions у нас появляется одно поле partition:

+----------+ 
|partition | 
+----------+ 
|2024-03-01| 
|2024-02-26| 
+----------+ 

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

#spark

💡Хинт NOLOCK

🌐Если в базу параллельно пишутся данные, а вам надо оттуда читать - nolock в помощь.

Как использовать?
Диалект - MS SQL Server

SELECT *
FROM test (NOLOCK)

🔐Что происходит?
NOLOCK убирает блокировки, связанные с параллельным обращением к одним и тем же данным. Другое название - READ UNCOMMITTED, т.е. нам не нужно ждать, пока транзакция завершится. Но тут возникает вопрос с чтением грязных данных - мы можем работать с данными, которые уже сто раз изменились.

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

#sql_tips

⤴️Оптимизация джойнов

В спарке существуют 2 вида трансформаций: узкие и широкие.

💃Узкие не требуют перемещений данных и на любом маленьком кусочке могут выполняться параллельно и независимо: where(), withColumn(), union(). Например, чтобы отфильтровать строки, нам не нужно знать весь датасет. Мы берем одну строку, применяем условие - готово.

🍊Широкие же требуют шафла: join(), groupBy(), sort(), distinct(). Здесь же нам нужен весь датасет. Допустим, мы хотим сделать дистинкт по полю color: на первом экзекьюторе лежат red, blue, green, на втором yellow, violet, blue. Если брать отдельно каждый экзекьютор, то цвета уникальны, но если мы возьмем все, то будут дубликаты. То есть нам сначала надо одинаковые значения собрать (это и есть шафл) и только потом почистить.

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

Есть несколько советов:
1️⃣Все фильтры до джойнов
2️⃣Использовать equi-джойны (SortMergeJoin, BroadcastHashJoin)
3️⃣Если можно увеличить данные, но вместо non-equi (NestedLoop, Cartesian) использовать equi, то делать именно так
4️⃣Если правый датасет помещается в память экзекьютора, использовать broadcast
5️⃣Избегать cross-join'ов
6️⃣Перепроверять в плане запросов

#spark

🎆Нас 400+!🎆

😉Всем подписчикам спасибо за то, что вы вкладываете свое время и читаете мои посты!

📝Я со своей стороны тоже стараюсь, чтобы вам было интересно и была практическая применимость всей инфы. Пару раз садилась писать на теоретические темы и в какой-то момент ловила себя на мысли, что пересказываю документацию, а этого не хотелось бы (или вам, наоборот, хотелось бы?).

📝Да, еще у меня есть бэклог с ~50 темами, которые я планирую раскрыть, так что все сказанное вами в комментах я аккуратно туда заношу)

В общем, этот пост для ваших идей, насущных вопросов и всего, чего хотите👇

🤗Подкапотные джойны в Spark

Все sql-джойны внутри превращаются в один из 4х типов:
➡️ BroadcastHashJoin
➡️ SortMergeJoin
➡️ BroadcastNestedLoopJoin
➡️ CartesianProduct

Какой из них используется - смотрим через df.explain()
Сейчас пройдемся по каждому и посмотрим на условия срабатывания.

1️⃣ SortMergeJoin - дефолтный

- условие на равенство


F.col('df1.id') == F.col('df2.id')

- ключи должны быть сортируемы
Примеры несортируемых ключей: бинарный формат, сложные структуры

2️⃣ BroadcastHashJoin

Что такое broadcast?

⬇️⬇️Как мы знаем, спарк нужен для параллельной обработки данных. При джойнах экзекьюторам приходится обмениваться данными (чтобы сопоставить одинаковые ключи), что приводит к операции shuffle - группируются одинаковые ключи из разных экзекьюторов на одном => большие расходы на сеть. Но если у нас есть маленькая табличка (десятки мегабайт, но по умолчанию просто 10), то мы можем скопировать ее на все экзекьюторы, поджойнить там же и избежать шафла. Лимит на размер - память самого экзекьютора.

- условие на равенство
- включен broadcast в настройках сессии

# 2й аргумент - размер в МБ
.config('spark.sql.autoBroadcastJoinThreshold', 100)

# так отключается broadcast
.config('spark.sql.autoBroadcastJoinThreshold', -1)

# broadcast join
df1.join(F.broadcast(df2), condition, join_type)

3️⃣ BroadcastNestedLoopJoin

- условие на неравенство
- включен broadcast

NestedLoop - потому что мы итерируемся по маленькому датасету и проверяем каждую строчку

4️⃣ CartesianProduct

- условие на неравенство
Самый медленный, может приводить к OOM-ошибкам (Out of Memory).

#spark

❌CROSS APPLY, OUTER APPLY

❌Я буду говорить про MS SQL Server (потому что очень долго с ним работала), но в Postgres тоже есть похожие конструкции - CROSS JOIN LATERAL, LEFT JOIN LATERAL. Так что с другими бд, я надеюсь, тоже несложно разобраться.

В чем суть?
По факту это те же самые джойны, но не с табличкой, а с результатом функции, которая возвращает табличку.

1️⃣ CROSS APPLY ~ INNER JOIN

Допустим, есть функция поиска книги по автору:

CREATE FUNCTION getBookByAuthorId(@AuthorId int)
RETURNS TABLE AS RETURN
( 
  SELECT * FROM book
  WHERE author_id = @AuthorId
)

Используем в запросе:

SELECT a.author_name, b.book_name, b.price
FROM author a
CROSS APPLY getBookByAuthorId(a.id) b

2️⃣ OUTER APPLY ~ LEFT JOIN

Ненайденные строчки также заполняются NULL.

В этом случае аналогично подзапросу в селекте:

--outer apply
SELECT a.author_name, book_list
FROM author a
OUTER APPLY (
  SELECT STRING_AGG(book_name, ', ') AS book_list
  FROM book b
  WHERE b.author_id = a.id
) sub

--подзапрос
SELECT a.author_name, (
  SELECT STRING_AGG(book_name, ', ')
  FROM book b
  WHERE b.author_id = a.id
) AS book_list
FROM author a
--STRING_AGG соединяет строки в одну ячейку с разделителем

С агрегирующими функциями прокатит только outer apply, потому что агрегировать поверх подзапроса нельзя. А так можно:

SELECT a.author_name, count(book_id) as book_num
FROM author a
OUTER APPLY (
  SELECT b.id as book_id
  FROM book b
  WHERE b.author_id = a.id
) sub
GROUP BY a.author_name

#sql_tips

📞Ребятушки, подъехала вакансия

Data Engineer в Яндекс 360
Москва/СПБ
middle, senior

Яндекс 360 — это Почта, Диск, Телемост с видеозвонками, Документы, Мессенджер, Рассылки, Заметки, Календарь, Трекер, Вики и Формы. Каждый день этими сервисами пользуются миллионы человек и тысячи компаний.
Наши продукты, команды и, конечно, объёмы данных быстро растут. Мы ищем дата-инженера, который поможет организовать хранилище, навести порядок и вывести аналитику наших сервисов на новый уровень.

Какие задачи вас ждут
- Вести разработку ETL-процессов поставки данных
- Выявлять неэффективность в существующих процессах и оптимизировать их
- Проектировать структуру хранения данных
- Погружаться в данные из прикладных бизнес-областей и становиться в них экспертом
- Сопровождать разработанные ETL-процессы
- Участвовать в развитии платформы DWH

Мы ждём, что вы
- Уверенно владеете Python
- Уверенно владеете SQL (join, агрегация, оконные функции, оптимизация сложных запросов)
- Умеете пользоваться Git
- Знаете, как строить и оптимизировать ETL-процессы

Будет плюсом, если вы
- Разбираетесь в слоях DWH и моделях хранения данных
- Работали с большими объёмами данных

Откликнуться

SQL джойны в Spark

🎹В спарке всего 7 видов:
1) inner
2) left outer
3) right outer
4) full outer
5) cross
6) left semi
7) left anti

На первых четырех останавливаться не будем, вы их точно знаете😉

Схема такая:
- с какой табличкой джойним
- условия джойна
- тип джойна

df = df1.join(df2, F.col('id') == F.col('parent_id'), 'left')

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

✨Часто бывает, что столбцы называются одинаково, тут в ход идут элиасы. Так мы не привязываемся к конкретным названиям датафреймов:

df = df1.alias('df1').join(df2.alias('df2'), F.col('df1.id') == F.col('df2.id'), 'left')

CROSS JOIN
В спарк конфиге при создании сессии есть параметр, который может включать и отключать cross join:

.config("spark.sql.crossJoin.enabled", True)

Если отключите, то в коде вызовется исключение. В Spark3 по умолчанию True, в Spark2 - False.

LEFT SEMI
То же самое, что inner join, но возвращаются только колонки из левого датасета для сметчившихся строк.

LEFT ANTI
Возвращает колонки из левого датасета, но для НЕсметчившихся строк.

🤗Вообще left semi я особо не использую, т.к. если уж нужно джойнить с другой табличкой, то наверняка оттуда надо что-то забрать😁 А вот left anti - удобная штука, когда надо поресерчить проблемные строки и не загромождать экран ненужными колонками

#spark