💡Ответ💡
Правильные варианты - 2, 4.

Тут есть несколько интересных моментов:

1️⃣Агрегация

Делается через .agg() - на вход подается функция, которая что-то считает:

F.min(), F.max(), F.count(), F.countDistinct(), F.sum(), F.sumDistinct(), F.avg(), F.mean и т.д.

Дополнительно в селекте прописывать не нужно:

df.groupBy("department") \
  .agg(F.sum("salary"), \
       F.avg("salary"))

2️⃣На любое поле можно навесить элиасы

df.groupBy("department")
  .agg(F.sum("salary").alias("sum_salary"), \
       F.avg("salary").alias("avg_salary"))

3️⃣ having не существует

После агрегации у нас появляется новое поле, по которому сразу можно фильтровать. Без элиасов поля будут называться как функция(поле)

# равнозначны
.agg(F.count("customer_id").alias("cnt"))
.where("cnt > 5")

.agg(F.count("customer_id"))
.where("count(customer_id) > 5")

4️⃣ where и filter взаимозаменяемы

Внутри можно использовать как стандартный sql в кавычках, так и делать питоновские/скаловские сравнения:

.where("date = '2024-02-08'")
.filter("date = '2024-02-08'")
.where(F.col('date') == '2024-02-08')
.filter(F.col('date') == '2024-02-08')

Обратите внимание, что на Scala тройное равно:

.where("date = '2024-02-08'")
.filter("date = '2024-02-08'")
.where(col("date") === "2024-02-08")
.filter(col("date") === "2024-02-08")

col - это обертка над колонкой, которая дает доступ к более крутым функциям поверх: alias(), desc(), like(), isNull() и т.д.

#spark

⚔️UNION vs CASE WHEN

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

Тогда я затупила, но у вас есть шансы пропустить этот этап😁
Они проверяли знание union и case when:

1️⃣ UNION/UNION ALL - для одного столбца

SELECT 'amt_not_null' as agg_type, SUM(amount) AS val
FROM test
WHERE id IS NOT NULL
UNION ALL
SELECT 'amt_null', SUM(amount)
FROM test
WHERE id IS NULL

UNION - удаляет дубликаты строк, UNION ALL - нет.
Если разницы для запроса нет, лучше выбирать UNION ALL, потому что без проверки на дубли он более производительный.
Элиасы во втором и последующем запросах можно не указывать, они будут тянуться из первого.

2️⃣ CASE WHEN - для разных столбцов

SELECT
  SUM(CASE WHEN id IS NOT NULL THEN amount ELSE 0 END) AS amt_not_null,
  SUM(CASE WHEN id IS NULL THEN amount ELSE 0 END) AS amt_null
FROM test

#sql_tips

💻Полезные консольные команды

Я частенько оставляю в заметках команды, которые попадаются не раз, тут списочек для терминала:

1️⃣ Настройки проекта

# путь к дефолтному питону, на котором все запускается
which python
# /c/Users/Username/AppData/Local/Programs/Python/Python311/python

# все пути к установленным питонам
where python
whereis python

# версия
spark-submit --version

# установить все зависимости из файлика
pip install -r requirements.txt

Вместо питона может быть java, mvn (maven - сборщик проектов) - все, что угодно.

2️⃣ GIT

# поменять сообщение последнего коммита
git commit --amend -m "new message"

# запуллить актуальную версию из УКАЗАННОЙ веточки репозитория и затереть все свои локальные изменения
git reset --hard origin/develop

# запуллить актуальную версию из ТЕКУЩЕЙ веточки репозитория и затереть все свои локальные изменения
git reset --hard HEAD

# удалить локальную веточку
git branch -d localBranchName

# удалить веточку в репозитории
git push origin --delete remoteBranchName

3️⃣ Linux

# найти файл по названию/расширению/паттерну в ТЕКУЩЕЙ директории
find . -name "*.txt"

# искать, начиная с корня
find / -name "*.txt"

# найти файлы в ТЕКУЩЕЙ директории, в которых содержится строка
grep -r "import" .

# искать, начиная с корня
grep -r "import" /

# заархивировать папку
zip -r file.zip source_folder

# разархивировать
unzip file.zip -d destination_folder

4️⃣ YARN

# убить приложение
yarn app -kill <app_id>

# найти, кто убил ваше приложение
yarn app -status <app_id> | grep killed

🍄Задачка

🌸Есть таблица с полем инн и кучей других полей. Столбец с инн может и будет дублироваться. Нам нужно прийти к структуре второй таблички.

occurrences - это сколько раз инн повторяется в таблице
num_inn - количество инн, которые встречаются occurrences раз

➡️Пример⬅️
156, 444, 777 - встречаются 1 раз, но всего их 3
283 - 2 раза, всего таких инн 1
900 - 4 раза, всего тоже 1

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

⭐А какие у вас идеи?⭐

💡Ответ💡

🦕Естественно, самый очевидный и оптимальный вариант использовать подзапрос:

SELECT
  occurrences,
  COUNT(1) AS num_inn
FROM (
  SELECT COUNT(1) AS occurrences
  FROM test
  GROUP BY inn
)
GROUP BY occurrences
ORDER BY num_inn DESC, occurrences;

Я обычно пишу count(1) вместо count(*), чтобы на всякий база не пошла читать все столбцы. Но многие говорят, что это миф и они абсолютно одинаковы, так что решать вам🪙

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

🍊 Второй вариант я никогда не применяла, но мы можем подать в оконку результат из group by. Это следует из порядка исполнения запроса:

from -> join -> where -> group by -> having -> select (вместе с оконками) -> order by -> limit

То есть на этапе селекта мы уже сделали все группировки и можем спокойно использовать:

SELECT DISTINCT
  COUNT(inn) AS occurrences,
  COUNT(COUNT(inn)) OVER (PARTITION BY COUNT(inn)) AS num_inn
FROM test
GROUP BY inn
ORDER BY num_inn DESC, occurrences;

📝Последняя строчка в плане для обоих запросов:

Sort(cost=43.09..43.59)

Unique(cost=52.31..53.81)

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

⬅️LEFT убивает индексы

Когда стоит вопрос: использовать LEFT или LIKE - надо брать второе. Особенно в базах, которые поддерживают индексацию в LIKE.

💳Пример - есть таблица с номерами карт, этот столбец индексирован, и мы хотим вытащить все MasterCard (начинаются на 5).

Какие есть варианты запроса?

SELECT * FROM accounts
WHERE LEFT(account_num, 1) = '5'

SELECT * FROM accounts
WHERE account_num LIKE '5%'

План запроса:

|--Clustered Index Scan(..., WHERE:(substring(account_num,(1),(1))=N'5'))

|--Clustered Index Seek(..., SEEK:(account_num >= N'5' AND account_num < N'6'), WHERE:(account_num like N'5%') ORDERED FORWARD)

🔫В первом случае мы убиваем навешенный индекс, потому что под капотом лежит Index Scan - вытаскиваем все строки из таблицы и ищем совпадение с подстрокой.

🔎Во втором случае под капотом уже Index Seek - сразу вытаскиваем нужную группу записей, потом фильтруем.

PROFIT👍

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

SELECT * FROM documents
WHERE YEAR(valid_from) = 2023 and MONTH(valid_from) = 6
--Index Scan 

SELECT * FROM documents
WHERE valid_from BETWEEN '2023-06-01' AND '2023-06-30'
--Index Seek 👍

#sql_tips

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

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

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

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

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

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

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

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

❌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

🤗Подкапотные джойны в 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

🎆Нас 400+!🎆

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

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

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

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

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

В спарке существуют 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

💡Хинт NOLOCK

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

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

SELECT *
FROM test (NOLOCK)

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

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

#sql_tips

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

➗Вообще партицирование/партиционирование - это метод разделения большой таблицы на маленькие кусочки для оптимизации обращения к ним. Часто партицируют по дате или по полю с небольшим конечным множеством значений. В 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

Работа с таблицами в 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://news.1rj.ru/str/rzv_de/132
А еще можете пройтись и по другим заметкам)

#hive

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

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

--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;

Опрос ниже👇

💡Ответ 💡

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

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

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

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

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

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

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

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

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

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

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