count(), F.count(), F.countDistinct(), distinct()

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

1️⃣ count() - это действие
Оно запустит весь пайплайн расчетов и выведет на экран одно число - количество строк в датафрейме:

df.count()

Также можно использовать в агрегации, но нельзя поменять название столбца (будет count):

df.groupBy('category')
  .count()
  .orderBy('count')
  .show()

2️⃣ F.count() - это трансформация, агрегирующая функция

df.select(F.count('id')).show()

В отличие от первого способа в агрегации можно поменять название столбца:

df.groupBy('category')
  .agg(F.count('id').alias('cnt'))
  .orderBy('cnt')
  .show()

Отличия count() и F.count():

# результат - датафрейм с одной ячейкой
df1.select(F.count('id')).show()
# +---------+
# |count(id)|
# +---------+
# |        3|
# +---------+

# результат - число
df1.select('id').count()
# 3

3️⃣ F.countDistinct() - тоже агрегирующая функция

df.select(F.countDistinct('id')).show()

Тут вариант агрегации только один, потому что df.countDistinct() не существует:

df.groupBy('category')
  .agg(F.countDistinct('id'))
  .show()

При этом в спарке есть еще функция distinct(), которая тоже вносит некую путаницу.

Отличия distinct().count() и F.countDistinct():

# результат - датафрейм с одной ячейкой
df1.select(F.countDistinct('id')).show()
# +------------------+
# |count(DISTINCT id)|
# +------------------+
# |                 2|
# +------------------+

# результат - число
df1.select('id').distinct().count()
# 2

#spark

Как из деления int/int получить double?

🤓Вам нужно что-то посчитать: поделить одно количество на другое и округлить результат до 2х знаков после запятой.

Так как количество - это целое число, если написать просто count1/count2, то в результате мы получим тоже целое число: 15/26 = 0. А мы хотим 0.57692307692307692308 - вот оно самое.

✨Поэтому умножаем на 1.0, и вуаля✨

SELECT round(count1 * 1.0 / count2, 2)
FROM test

#sql_tips

👍Полезные советы

В продолжение поста про вредные советы есть еще несколько лайфхаков (уже нормальных!):

1️⃣ Бывает полезно сделать такое форматирование, когда мы играемся с полями, чтобы удобнее закомментить ненужные:

SELECT 
    e.emp_no
    , e.first_name
    --, e.last_name
    , s.salary
FROM employees e
  JOIN salaries s
    ON e.emp_no = s.emp_no

2️⃣ Добавлять элиасы таблиц к полям (sql или spark - неважно), чтобы под конец огромного запроса понимать, откуда тянутся данные. То же самое 👆на спарке:

final_df = (
  employees.alias('e')
  .join(salaries.alias('s'), F.col('e.emp_no') == F.col('s.emp_no'), 'inner')
  .select(
    'e.emp_no',
    'e.first_name',
    's.salary'
  )
)

3️⃣ А когда мы играемся с условиями, я частенько добавляю 1=1, чтобы тоже быстро закомментить и прогнать:

SELECT *
FROM employees
WHERE 1=1
     --AND date_joined <= '2024-01-01'
     AND is_deleted = 0
     AND username != 'system'

Если бы WHERE date_joined <= '2024-01-01' шло на первом месте, то пришлось бы убирать AND перед вторым условием, удалять лишние пробелы - в общем, неудобно, если часто так делать.

#sql_tips #spark

Задачка

Есть 3 таблички:

1️⃣ Cities (id, city, region)
2️⃣ Couriers (id, fio)
3️⃣ Orders (id, datetime, cost, city_id, courier_id)

Найти топ 10 курьеров для каждого города, которые доставили больше всего заказов за март.

Помогите закрыть таску!😘

💻Исходные данные
У нас есть три вьюшки с id и json-полем. Вьюшки собираются по разным фильтрам.

📌Цель: собрать массив степов, при этом дополнить предыдущие недостающие степы значением default_json.

🎥Пример:
для id = 1: [json]
для id = 2: [default_json, json]
для id = 3: [default_json, default_json, json]
и т.д.

При этом в default_json прокидывается текущий степ, поэтому нам нужно не потерять это название.

🚀Пока идея такая:
- делаем full join
- делаем unpivot
- добавляем флаг is_missed на основе null
- итерируемся с конца в рамках id и удаляем все is_missed = 1, пока не найдём is_missed = 0

Лично мне это кажется слишком громоздкий и малочитаемым, поэтому хочется двинуться в сторону unuon, но тогда теряется текущий степ. Или после джойна в рамках строки как-то найти последнее заполненное поле.

🤩Any ideas?

Логика с json'ами

На прошлой неделе я вкидывала задачу. Мы ещё раз пообсуждали и вот на чем остановились:

1️⃣Делаем фулл джойн исходных витрин: view1, view1_1, ...
Идею с кросс джойнами мы решили оставить)

2️⃣Делим фулловую таблицу на n вьюшек, где n - это количество шагов. Начинаем с конца (последнего заполненного).

Пример:
when step1_2 is not null => v_step1_2 - новая вьюшка для строк, у которых есть степ1_2:

3 null null json
5 json null json
6 null json json
7 json json json

3️⃣Делаем left anti join фулловой таблицы с созданной вьюшкой, получаем строки, которые не вошли:

1 json null null
2 null json null
4 json json null

Повторяем шаги 2 и 3 n раз.
NB! В следующих вьюшках количество столбцов уменьшается (исключаем уже выделенный степ)

1) when step1_1 is not null => v_step1_1

2 null json
4 json json

После left anti join обновленная фулловая:

1 json null null

На последнем этапе просто берем id и первый столбец.

Итак, у нас 3 вьюшки с разным количеством столбцов.

️4️⃣Делаем coalesce(stepn, default_json)

5️⃣Делаем array(всех шагов) => получаем одинаковую структуру для всех вьюшек с полями id и array

6️⃣Union

🕺🕺🕺🕺🕺🕺🕺

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

📖 Наконец-то прочитала книжку "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. все, идите отдыхайте 😶🐱🐱