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

Объявление переменных 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

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

Бывает так, что есть поле, по которому нужно отсортировать, но по алфавиту ну никак не подходит. Тогда можно искусственно создать поле для сортировки через 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

🔗Обязательные импорты в 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

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

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

⚙️Основные настройки 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

💡Ответ💡
Верные варианты: 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

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

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

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

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

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

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

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

CREATE TABLE #localTempTable (...)

CREATE TABLE ##globalTempTable (...)

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

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

🔖Полезные книжки по спарку

2020
Learning Spark v2
Spark в действии
Data Analysis with Python and PySpark

2018
Spark: The Definitive Guide

2017
Эффективный Spark. Масштабирование и оптимизация

Ссылочки
The Internals of Spark SQL
Mastering Apache Spark 2.0
How Query Engines Work

Поговорим про ✨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

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

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

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

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

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

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'

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

🤭Санте нужна твоя помощь!🤭

🛷У Санта Клауса есть кучка писем, в которых дети просят самую желанную игрушку. Некоторые из них заказали хрупкие подарки, которые могут сломаться, а Санта Клаус ну никак не хочет испортить праздник❔ Он подумал, что электронной технике и музыкальным инструментам стоит уделить особое внимание.

👣Санта Клаус у нас современный и хочет учесть логистику. Он придумал, что эти подарки купит прямо в стране ребенка, а не привезет с собой.

Помогите Санта Клаусу составить список, какие игрушки и в каких странах нужно будет докупить до нового года!🙏

letters (id, date, child_id, toy_id)
children (id, name, country_id)
country (id, name)
toy (id, name, category)

И всех с наступающим!👋
Добра, вдохновения, мотивации, удачи, сил и прекрасного настроения!🥳

❄️Ответ на задачу 2❄️

Один из вариантов был таким:

SELECT DISTINCT t1.*
FROM logs t1
  JOIN logs t2
    ON t1.id > t2.id AND t1.dt < t2.dt;

⚠️Но посмотрим на план запроса (читаем снизу вверх, смотрим на cost):

HashAggregate (cost=92291..92294) 
  Group Key: t1.id, t1.dt
  -> Nested Loop (cost=0..89454)
    Join Filter
    -> Seq Scan on logs t1 (cost=0..33)
    -> Materialize
      -> Seq Scan on logs t2

Здесь очень дорогой Nested Loop Join, который увеличил косты с 33 до 90к.

✅Что ожидалось увидеть?

Используем lag/lead и сравниваем разницу айдишников с предыдущим и последующим:

WITH diffs AS ( 
  SELECT 
    *,
    id - LAG(id) OVER(ORDER BY dt) prev_diff,
    id - LEAD(id) OVER(ORDER BY dt) next_diff
  FROM logs
) 
SELECT id, dt
FROM diffs
WHERE prev_diff > 1 or next_diff > 1;

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

Subquery Scan on diffs (cost=159..249)
  Filter
  -> WindowAgg
    -> Sort
      Sort Key: logs.dt
      -> Seq Scan on logs

В первом случае примерные косты были 90к, во втором 250 => в 370 раз меньше.

✨Также нам необязательно знать все id поздних записей, достаточно найти границы диапазонов✨

❄️Ответ на задачу 1❄️

😎Ребята в комментариях дали правильные ответы (круть!), тут предложу чуть оптимизированные вариации:

1) JOIN

Один подзапрос + один джойн

SELECT c.client_id
FROM clients c
LEFT JOIN (
    SELECT DISTINCT client_id
    FROM documents
    WHERE valid_to = '9999-12-31'
) t1
  ON c.client_id = t1.client_id
WHERE t1.client_id IS NULL

2) IN

Идём от обратного и исключаем

SELECT client_id
FROM clients
WHERE client_id NOT IN (
    SELECT client_id
    FROM documents
    WHERE valid_to = '9999-12-31'
)

3) NOT EXISTS

Здесь можно оптимизировать, используя top/limit и константу.

Как это работает?
Если условие подзапроса соблюдается, мы берем просто цифру 1 вместо полей (не нужно лишний раз их вытаскивать) + одной строки нам достаточно, чтобы понять, что данные есть или их нет.

SELECT client_id
FROM clients c
WHERE NOT EXISTS (
    SELECT 1
    FROM documents d
    WHERE c.client_id = d.client_id
      AND valid_to = '9999-12-31'
    LIMIT 1
)

Также важно внимательно читать задание и выводить то, что требуется (только id клиентов).

Решившим ребятам респект, остальные подключайтесь тоже!

Задача 2.

Есть события, но часть из них доезжает позже других.
Нужно вывести такие события (5 и 6, которые вклинились).

create table logs (
  id int,
  dt date
);

insert into logs values
(1, '2023-12-01'),
(2, '2023-12-01'),
(5, '2023-12-02'),
(6, '2023-12-03'),
(3, '2023-12-06'),
(4, '2023-12-08');