🎉Все подарки доставлены, Санта благодарит за спасение нового года!

Итак, вариант от спасителя:

SELECT t.name, c.name, COUNT(*) AS count_toys
FROM country AS c
  JOIN children AS ch
    ON ch.country_id = c.id
  JOIN letters as l
    ON l.child_id = ch.id
  JOIN toy as t
    ON t.id = l.toy_id
WHERE l.date > '2023-01-01'
GROUP BY t.name, c.name

Обращаю внимание на один момент - фильтр на дату можно перенести повыше:

JOIN letters as l
  ON l.child_id = ch.id
  AND l.date > '2023-01-01'

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

😉Коллеги нынешние, прошлые и будущие!

Написала свою первую статью на хабре про то, как я анализирую свои доходы и расходы. Может, и вам будет полезно☕️

https://habr.com/ru/articles/784412/

Поговорим про ✨Apache Spark✨ - это движок/фреймворк для распределенной обработки больших данных.

Что значит распределенной?
Представь, что ты археолог и тебе нужно раскопать огромную территорию. Ты решил позвать n друзей, вы начали работать параллельно, и теперь вы закончите в n раз быстрее. Так и в спарке: каждая операция делится на маленькие таски, которые одновременно обрабатываются несколькими компьютерами, что ускоряет весь процесс.

Со спарком обычно работают на Python (через либу PySpark) и Scala.

Сначала нужно создать SparkSession:

from pyspark.sql import SparkSession

spark = (SparkSession.builder
  .appName("SparkExample")
  .master("yarn")
  .config("spark.some.config.option", "config-value")
  .enableHiveSupport()
  .getOrCreate()
)

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("SparkExample")
  .master("yarn")
  .config("spark.some.config.option", "config-value")
  .enableHiveSupport()
  .getOrCreate()

Пару слов про code style в питоне. Есть два варианта:

1) обратный слэш

spark = SparkSession.builder \
  .appName("SparkExample") \
  ...

2) скобки

spark = (SparkSession.builder
  .appName("SparkExample")
  ...
)

Я лично предпочитаю второй подход, потому что в первом нельзя закомментить строчки (только удалить, иначе синтаксическая ошибка) и нужно проставлять бэкслэши на каждой строке. Во втором - только один раз обрамить скобками, и все👌

#spark

📌Помимо #spark, также буду вести рубрики #sql_tips, #python_tips, возможно, что-то еще по мере появления идей. Там будут лайфхаки, о которых я узнала на рабочих задачах или во время чтения книжек📖

Например, про временные таблицы⏳

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

В MS SQL Server есть два вида: локальные (#) и глобальные (##):

CREATE TABLE #localTempTable (...)

CREATE TABLE ##globalTempTable (...)

Локальные доступны только для вашего пользователя и удаляются при закрытии сессии. Глобальные доступны всем и живут, пока жива хотя бы одна использующая их сессия.

На временные таблицы также можно навешивать индексы🗂

💡Ответ💡
Все варианты валидны.

В сортировке можно использовать:

1) оригинальные поля

ORDER BY column_name

2) вычисляемые поля

ORDER BY fn(column_name)

3) элиасы (псевдонимы) полей

SELECT column_name AS column_name_alias
...
ORDER BY column_name_alias

4) числовые индексы полей (начиная с 1)

ORDER BY 1

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

А вам какой способ больше нравится?

#sql_tips

💡Ответ💡
Верные варианты: 6, 7, 8, 9, 10.

Т.к. в селекте два поля таблицы, то они оба должны фигурировать в группировке (если у нас нет никаких ухищрений). Сама агрегирующая функция (count) никогда там не указывается.
В зависимости от диалекта некоторые варианты работать не будут (выделены ❕).

В группировке можно использовать:
1) оригинальные поля

GROUP BY column_name

2) вычисляемые поля

GROUP BY fn(column_name)

3) элиасы (псевдонимы) полей❕

SELECT column_name AS column_name_alias
...
GROUP BY column_name_alias

В Postgres, Clickhouse прокатит, а в MS SQL Server уже нет.

4) числовые индексы полей (начиная с 1)❕

GROUP BY 1

Аналогично

5) оригинальные поля в вычислениях

SELECT fn(column_name)
...
GROUP BY column_name

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

Колонки и индексы можно комбинировать, но как по мне - лучше придерживаться одного стиля, иначе получается какая-то каша🥣

#sql_tips

⚙️Основные настройки Spark Session⚙️

Spark-приложение управляется менеджером ресурсов YARN/Mesos/Standalone. У нас используется YARN. Он запускает приложение, выделяет ресурсы на вычисления и мониторит весь процесс.

➡️Есть две важные вещи: driver и executor.

Executor - исполнитель, выполняет Spark-код (например, "выведи 10 строчек из таблицы"). Их много.
Driver - драйвер, координирует работу экзекьюторов, планирует для них задачки и собирает результаты. А он такой один.

📝Перейдем к параметрам:

spark.driver.memory - объем памяти драйвера
spark.driver.cores - количество ядер драйвера
spark.driver.maxResultSize - максимальный размер результата, который передается от экзекьютеров к драйверу после вычислений

spark.executor.memory - объем памяти одного экзекьютора
spark.executor.cores - количество ядер экзекьютора
spark.executor.instances - количество экзекьюторов
Количество экзекьюторов и ядер влияет на скорость обработки данных за счет параллельного вычисления.

spark.local.dir - директория хранения временных файлов
spark.port.maxRetries - максимальное количество попыток подключения к порту (для UI, драйвера и т.д.)

Как применять:

spark = (
  SparkSession.builder
  .config("spark.driver.memory", "20g")  # g - гигабайты, возможны 4: k, m, g, t
  .config("spark.driver.cores", "2")
  .config("spark.driver.maxResultSize", "20g")
  .config("spark.executor.memory", "10g")
  .config("spark.executor.cores", "2")
  .config("spark.executor.instances", "20")
  .config("spark.local.dir", "sparktmp")
  .config("spark.port.maxRetries", "150")
  ...
)

Для запуска у себя на ноуте достаточно:

spark = (
  SparkSession.builder
  .appName("MySparkApp")
  .master("local[*]")  # * - все ядра, или указать число
  .getOrCreate()
)

Более подробный список тут

Экзекьюторы нужно настраивать при статическом выделении ресурсов, а как это правильно делать и как настраивать динамически - расскажу в следующих постах👨‍💻

#spark

🕺Динамическое выделение ресурсов

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

📝Перейдем к параметрам:

spark.dynamicAllocation.enabled - флаг динамического/статического выделение ресурсов

spark.dynamicAllocation.initialExecutors - начальное количество экзекьюторов при старте

spark.dynamicAllocation.minExecutors - минимальное количество экзекьюторов

spark.dynamicAllocation.maxExecutors - максимальное количество экзекьюторов

spark.dynamicAllocation.executorIdleTimeout - время экзекьютора на ничегонеделание, по истечении которого он будет удален

spark.dynamicAllocation.cachedExecutorIdleTimeout - время экзекьютора с закешированными данными на ничегонеделание, по истечении которого он будет удален

Закешированные данные - когда результат вычисления хранится в памяти. Т.е. есть датафрейм, который что-то джойнит. Вы подождали 10 минут, пока посчитается, закешировали. В следующий раз обратились к тем же данным - и подождали уже 2 секунды.

Как применять:

spark = (
  SparkSession.builder
  .config("spark.dynamicAllocation.enabled", "true")
  .config("spark.dynamicAllocation.initialExecutors", "2")
  .config("spark.dynamicAllocation.minExecutors", "0")
  .config("spark.dynamicAllocation.maxExecutors", "5")
  .config("spark.dynamicAllocation.executorIdleTimeout", "600s")  # 10 минут
  .config("spark.dynamicAllocation.cachedExecutorIdleTimeout", "600s")
  ...
)

Когда кластер свободный, а вам надо обработать много данных, то это удобный способ. Но если вы долго ничего не считаете и таймауты из конфигов уже прошли, то у вас останется minExecutors. И если за это время кластер займут, то вы уже не сможете полноценно работать, пока не попросите коллег пожертвовать своими ресурсами🙏

#spark

👻Про явное приведение типов

1️⃣ При работе со строками в sql для экономии памяти лучше ограничивать максимальную длину поля: varchar(100). В MS SQL Server также есть возможность указать varchar(max). Она расширяет количество символов, но тогда на это поле нельзя навесить индексы.

2️⃣ Если мы хотим поджойнить две таблицы, но поля имеют разные типы, то нужно кастить одно из них:

SELECT *
FROM test1
  JOIN test2
    ON cast(test1.id as varchar) = test2.id

или

    ON test1.id = cast(test2.id as int)

Некоторые бд сами подсвечивают ошибку при запуске запроса, а некоторые начинают его выполнение, но затрачивают сильно больше ресурсов на неявное преобразование. И запрос, который мог отработать за 5 минут, может отработать за 30 или не отработать вообще. Поэтому за соответствием типов важно следить🤓

#sql_tips

🔗Обязательные импорты в Spark-приложении

# сессия
from pyspark.sql import SparkSession
# функции
from pyspark.sql import functions as F
# типы данных
from pyspark.sql import types as T
# оконки
from pyspark.sql.window import Window

F и T - это code-style, принятый в PySpark, чтобы избежать пересечений с другими либами. В коде будет так: F.function(args).
И вообще импортируем только то, что нужно. import * - это моветон.

// датафрейм и сессия
import org.apache.spark.sql.{DataFrame, SparkSession}
// функции
import org.apache.spark.sql.functions._  // импорт всего
// udf (кастомные функции) и оконки
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
// типы данных
import org.apache.spark.sql.types._

В отличие от питона, в скале нужно указывать типы аргументов в функциях, поэтому мы дополнительно импортируем DataFrame, UserDefinedFunction и Window, т.к. они наиболее часто используются. А сами оконки лежат в модуле functions.

def func(df: DataFrame, time_window: Window): DataFrame = {...}

#spark

🫡Кастомная сортировка

Бывает так, что есть поле, по которому нужно отсортировать, но по алфавиту ну никак не подходит. Тогда можно искусственно создать поле для сортировки через case when и его аналоги:

SELECT color
FROM test
ORDER BY
  CASE color
    WHEN 'red' THEN 1
    WHEN 'orange' THEN 2
    WHEN 'yellow' THEN 3
    WHEN 'green' THEN 4
    WHEN 'blue' THEN 5
    WHEN 'indigo' THEN 6
    WHEN 'violet' THEN 7
  END

#sql_tips

Объявление переменных Python🐍 vs Scala🏔

Если мы хотим объявить несколько переменных в одной строке, на Python это будет так:

a, b = 'a', 1.1

На Scala нужно добавить ключевое слово val/var и обрамить скобочками:

val (a, b) = ("a", 1.1)

Для удобства можно указать типы полей:

val (a:String, b:Double) = ("a", 1.1)

val - неизменяемый, от слова "value"
var - изменяемый, от слова "variable"

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

В Scala есть еще одна интересность - lazy val.
val выполняется, когда мы создали переменную.
lazy val выполняется, когда мы впервые обратились к этой переменной.
Если мы никогда до нее не доберемся, то ресурсы тратится не будут💳

#python_tips #scala_tips

❔Вопрос на подумать

Почему вчера запрос отрабатывал 5 минут, а сегодня 2 часа?

Так почему запрос стал медленнее?

Это вопрос с собесов, и тут главное - не перечислить все варианты, а показать, что вы умняши и соображаете🤓

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

💬Очень полезные варианты - в комментариях к предыдущему посту.

🧐И, прежде чем столько ждать, лучше провести свой анализ, посмотреть на каунты, план запроса, спросить у коллег. Иначе 2 часа могут растянуться и на подольше🥺

#собес

🎙Задачка с собеса

Смотрим на картинку.
В первой табличке хранятся приходящие события в разбивке по типу. Во второй - что мы должны получить в итоге.

🛍Представим себе, что это интернет-магазин. Пусть событие 1 - это клик на кнопку "Каталог", а событие 2 - это "Добавить в корзину". Т.е. на нашем сайте сначала 4 раза кликнули по каталогу, потом 3 раза положили в корзину и т.д. Разные дни - не суть важно.

Теперь мы хотим определить, когда каждое из типов событий началось, а когда закончилось. К примеру, 5го января пошло событие 2, и мы считаем, что это конец события 1.

❔Какие идеи?

#собес

💡Ответ на задачку💡

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

Я предлагаю такое решение:

SELECT
  event_type,
  date AS date_start,
  LEAD(date) OVER(ORDER BY date) AS date_end
FROM (
  SELECT
    event_type,
    date,
    LAG(event_type, 1, -1) OVER(ORDER BY date) AS prev_event_type
  FROM events) t
WHERE event_type != prev_event_type;

Что происходит?

1️⃣ Подтягиваем предыдущий тип события с офсетом и значением по умолчанию.

LAG - предыдущий элемент, LEAD - последующий.
offset = 1 - берем предыдущий элемент, c 2 был бы предпредыдущий и т.д.
default_value = -1 - применится для самой первой строки, т.к. еще не с чем сравнивать. Отрицательные события вряд ли будут, поэтому значение кажется безопасным для любой выборки.

2️⃣ Оставляем только те строки, где текущее событие не равно предыдущему.

После этого этапа в данных лежит:

event_type  date        prev_event_type
1           2024-01-01  -1
2           2024-01-05  1
1           2024-01-08  2
...

3️⃣ Остается только подтянуть следующую дату по порядку.

Готово!👍

Как считать датасет в Spark?

Все, мы импортировали все, что нужно, создали сессию, нашли датасет, готовы работать!

Самые популярные способы:
⭐ csv

df = (
  spark.read
  .option('header', True)
  .option('delimiter', ';')
  .option('inferSchema', False)
  .csv(path)
)

val df = spark.read
  .option("header", "true")
  .option("delimiter", ";")
  .option("inferSchema", "false")
  .csv(path)

На месте path может быть 'file.csv', несколько файлов 'file1.csv,file2.csv' или вся папка целиком 'myfiles/csv_data'.

Из опций основные:

🐱 header - заголовок
🐱 delimiter - разделитель полей
🐱 inferSchema - автоматический определитель типов полей

На больших данных использовать inferSchema опасно!

Почему? Потому что спарк пойдет сканировать каждую строчку, чтобы определить корректный тип. Допустим, есть колонка age, заполненная числами типа int, но в сааамом конце у нас все поехало и попало число типа double. В конечном итоге столбец будет тоже иметь тип double, но для этого мы просканировали все на всякий случай. И так с каждым полем. Ну такое😕

Поэтому надо самим задать схему и указать ее при чтении файла:

table_schema = StructType([
  StructField('name', StringType(), True),  # True - допустимы ли null
  StructField('age', IntegerType(), True),
  ...
])

df = (
  spark.read
  .option(...)
  .schema(table_schema)
  .csv(path)
)

StructType - класс для структуры датафрейма. StructField - для поля. Лежат в spark.sql.types.

⭐json, parquet, hive-таблица

df = spark.read.json(path)
df = spark.read.parquet(path)
df = spark.table('db_schema.table_name')

У нас есть HDFS (Hadoop Distributed File System) - это распределенная файловая система, в которой обычно хранятся большие данные. Там они лежат в виде файликов внутри папочек. А Hive - это обертка, чтобы мы могли читать данные из файликов через sql-запросы🐝

#spark