Как Spark оптимизирует запросы

Spark вроде бы понятный — прочитал parquet, отфильтровал по статусу, посчитал count — всё просто. Но иногда одну и ту же команду можно написать так, что отработает за секунду, а в другом случае будет работать часы.
Почему? Давайте разберемся
Spark не магическая коробка, которая непонятно как молотит сотни гигабайт данных, а использует конкретные хорошо известные нам техники.
Я бы хотел остановиться на трех из них и показать, как не сломать их

1. Predicate pushdown
Spark может передавать фильтры (WHERE) сразу в источник — parquet, ORC, JDBC. То есть не грузить весь файл, а сразу читать только нужные строки.
Пример:

df = spark.read.parquet("s3://...").filter("status = 'active'")

Когда работает:
— фильтр простой (=, >, <)
— источник поддерживает pushdown (паркет, ORC, JDBC)
— фильтр указан явно, а не через UDF

Когда не работает:
—фильтр через UDF
—фильтр через выражение (df["a"] + df["b"] > 5)
—JSON и CSV (там pushdown нет вообще)
Проверить можно через план запроса df.explain(True). Ищите PushedFilters.

2. Column pruning
Если тебе нужны только user_id и amount, Spark может прочитать только эти поля. Особенно важно, если таблица на 300+ колонок.
Пример:

df = df.select("user_id", "amount")

Когда работает:
— n.select(...), без *
— формат — parquet, ORC, JDBC
Когда не работает:
— select("*")
— ты передал df в .rdd.map(...), .apply(...) или UDF
Поэтому, когда работаете с большими данными, рекомендую первое время вообще не использовать SELECT *, потому что в таблице может быть 100/200/100500 колонок, и тянуть их все влечет проблемы с производительностью. Вряд ли вам нужны все сто колонок
В explain будет Scan parquet [user_id, amount], если всё хорошо.

3. Partition pruning
Последнее по списку, но не по значимости
Если данные разложены по папкам (event_date=...), Spark может читать только нужные партиции, а не все подряд.
Пример:

df = spark.read.parquet("s3://events/").filter("event_date = '2025-07-30'")

аналогично через Spark SQL

spark.sql("SELECT id, date FROM table WHERE date = '20250801'")

Когда работает:
Очевидно ,когда таблица партицированна
Когда не работает:
— Фильтр через UDF или через функцию, например: year(event_time)
— Spark (да и не только) не умеет применять оптимизации к колонкам внутри функций
— JSON/CSV без схемы и partition discovery
Проверяется через PartitionFilters в explain(True).

Итак, точно ломают Spark-оптимизации:
— select("*"). Всегда.
— отсутствие фильтра по колонке партицирования
— функции на колонках, по которым идет фильтр
— UDF. В целом их можно использовать, но очень аккуратно и желательно стараться обходиться стандартными средствами
— использование формата вроде csv, json и т.д.

Так что следуйте этим простым правилам, и Spark будет у вас летать 😎
Ну и конечно всегда открывайте и читайте план запроса!

Подписывайтесь на канал, ставьте огонечки, будем и дальше разбираться с технологиями в Big Data! ⚡️
Также записывайтесь на наш курс по инженерии данных, чтобы стать крутым DE 🔥

@bigdata_postupashki

Как устроен Trino — разбор для дата-инженеров

Наверняка многие из вас слышали про эту технологию в последнее время, её часто видно на митапах, поэтому давайте разберемся, почему же Trino сейчас становится таким популярным?
Trino (раньше назывался PrestoSQL) — федеративный SQL-движок для запросов к разным источникам: S3, Iceberg, Hive, PostgreSQL, Kafka и др. Он не хранит данные, а исполняет запросы напрямую — туда, где данные лежат. Сейчас часто используют его именно в Lakehouse архитектуре, как основной инструмент взаимодействия с данными.

Давайте же разберемся, из чего он состоит.

Coordinator — планирует запросы
Главный процесс в Trino: принимает SQL-запрос, строит логический и физический план, применяет оптимизации (pushdown, partition pruning, join-стратегии), разбивает план на фрагменты и распределяет их воркерам
Coordinator чем-то похож на Spark Driver — но проще: не участвует в shuffle, не управляет состоянием, не держит данные. Его задача — распланировать и отдать.

Workers — исполняют задачи
Воркеры получают части плана и исполняют их: читают данные, фильтруют, джойнят, считают. Stateless — не кэшируют и не сохраняют контекст между запросами. Просто получили задачу, сделали, отдали результат.
В отличие от executors в Spark, Trino workers не привязаны к приложению или сессии. Один запрос — одна жизнь.

Catalogs — подключение к источникам
Trino сам по себе ничего не знает про таблицы. Всё взаимодействие с данными идёт через каталоги. Каталог — это конфигурация, в которой описано, где лежат данные, как их читать и какие метаданные использовать (HMS, Iceberg REST, JDBC и т. д.).
Это одно из ключевых отличий от Spark. В Spark чаще всего используется один Hive Metastore как основной каталог, а подключения к внешним базам делаются отдельно — через JDBC внутри SparkSession. В Trino — наоборот: всё подключается через каталоги. Хочешь работать с PostgreSQL? Подключаешь каталог postgres.properties. Нужно S3 с Iceberg? Заводишь каталог iceberg.properties. Trino обрабатывает всё единообразно.

Пример:

SELECT o.user_id, o.order_id, u.region
FROM postgres.sales.orders o
JOIN iceberg.analytics.users u 
  ON o.user_id = u.user_id
WHERE u.region = 'eu-west'

Здесь Trino:
1. читает таблицу orders из PostgreSQL
2. читает таблицу users в формате iceberg в S3
3. применяет фильтр (Trino умеет передавать фильтры в источники данных)
5. делает join
6. возвращает результат
Это одна из главных причин использовать Trino. В Spark такое возможно, но требует ручного объединения DataFrame’ов, подключения JDBC, прописывания схем и довольно много обвязки.
Кроме того, он прекрасно интегрирован с Iceberg(как и Spark): умеет не только читать данные, но и модифицировать их — в связке с Iceberg доступны DELETE, MERGE, UPDATE.
Поэтому Trino стоит использовать как единую точку входа в ваш lakehouse, а также инструмент для построения BI-отчетов и дашбордов (он хорошо работает с Superset, Metabase и другими)

Тrino + Iceberg это сейчас довольно модная связка, которую обкатывают многие компании, и сами по себе технологии активно развиваются (можно хотя бы по активности на GitHub удостовериться). Поэтому рекомендую поизучать, ведь она все чаще встречается в архитектуре и всплывает на собеседованиях.
Полезно и в теории, и в проде 🤓

А также рекомендую подписаться на наш канал, чтобы первыми узнавать про всё самое интересное в Big Data 😎

Ставьте ❤️ если хотите еще разборов технологий
Ставьте 🔥 если уже работали с Trino

Также записывайтесь на наш курс по инженерии данных, чтобы стать крутым DE 🔥

@bigdata_postupashki

Задача с собеседования в Спортмастер
В ней проверяют не столько умение решать задачи на sql, сколько ваше умение использовать df api в Spark — по тому, как вы решаете, сразу можно понять, сколько вы знаете про Spark.

Условие:
Даны таблицы:
sales(datetime, shop, art, quantity) — продажи товаров (quantity может быть отрицательным — возвраты).
prices(art, price) — цены товаров.

Задача:
Найти выручку магазина 100 за 2013-01-01.
Выручка = quantity * price.
Возвраты учитывать.

Пример данных:
Sales
datetime | shop|art| quantity
01-01-13 12:12:12 | 100 |A1 | -2
01-01-13 12:12:13 | 100 |A1 | 3
Prices
art|price
A1 |10
Собственно, решение тут совсем несложное, как на SQL, так и на Spark DF, поэтому рекомендую сначала самому решить быстро, а потом проверить 🤓

Решение:
SQL (Spark SQL)
SELECT SUM(s.quantity * p.price) AS revenue
FROM sales s
JOIN prices p USING (art)
WHERE s.shop = 100
AND to_date(s.datetime) = '2013-01-01';

PySpark DataFrame API
from pyspark.sql.functions import col, to_date, lit, sum as sum_
# сначала не забываем делать фильтры и селекты, а только после джоины и агрегации
sales = spark.read.table("sales")\
.withColumn("date", to_date("datetime"))\
.filter((col("shop") == 100) & (col("date") == lit("2013-01-01")))\
.select("art", "quantity")

prices = spark.read.table("prices").select("art", "price")

df = sales.join(prices, "art", "inner")\
.withColumn("amount", col("quantity") * col("price"))\
.agg(sum_("amount").alias("revenue"))

А чтобы с легкостью решать такие задачи рекомендую записывать на наш курс по дата инженерии 😎

@bigdata_postupashki

SQL на стажировку в Т-банк

Дедлайн 26 августа. Ответы на тест по SQL, условие, а ответы ниже:

1. FROM/JOIN, WHERE, GROUP BY, HAVING, WINDOW, SELECT, ORDER BY, LIMIT/OFFSET
2. LEFT JOIN
3. COUNT, SUM, AVG
4. Возвращает первое ненулевое значение из списка
5. Не допускает NULL и определяет уникальность строк
6. DROP TABLE
7. NULL
8. LIKE "__X%" (два подчеркивания)
9. Нельзя использовать sum агрегат в where -> нужно перенести в Having
10. Подзапрос возвращает тысячи строк

А разбор задач по SQL уже на нашем курсе дата инженер и аналитика хард, на которые скидка 50% только до 26 августа.

@bigdata_postupashki

Почему в Big Data так часто используют HDFS, Hive и Spark?

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

Этапы работы стека

🏠 HDFS — хранение
Файловая система, принцип работы которой простой: есть главная нода (Namenode) с метаинформацией — названием файлов, их расположением и т.д., и подчиненные (Datanode), где непосредственно размещаются файлы.
Файлы (CSV, JSON, Parquet, ORC) разбиваются на блоки и распределяются по узлам кластера. Есть репликация для защиты от падений. Очень просто масштабируется и есть возможность хранить огромные объёмы (терабайты и петабайты).

🔍 Hive — доступ к данным
Изначально появился как движок, который бы позволял писать SQL для запросов к данным, лежащим в HDFS, без необходимости писать Java-приложения.
Есть такая полезная вещь, как Hive Metastore, которая хранит метаданные и можно использовать даже без непосредственно Hive (но об этом в следующий раз, когда будем говорить про Iceberg).
Довольно легко прикручивается к BI-инструментам .

⚙️ Spark — вычисления
Нам всем знакомый уже движок для обработки больших объёмов. Поддерживает batch, streaming, ML.
Он нативно умеет читать данные напрямую из HDFS или через Hive.
Работает в памяти, поэтому быстрее классического MapReduce.
Можно строить ETL-пайплайны только с помощью него — умеет читать по jdbc напрямую из разных баз и складывать в хранилище, и более того — может это параллелить в несколько потоков

Плюсы и минусы связки

✅ Проверенное решение: десятки лет используется в продакшне. Практически везде вы найдете Hadoop
✅ Гибкость: можно хранить любые форматы и обрабатывать разными инструментами.
✅ Подходит для разнообразных сценариев: от ETL до аналитики и ML.

❌ Не всегда оптимально для ad-hoc аналитики: Hive довольно медленный, поэтому для быстрых запросов лучше подойдут Trino/Impala или же отдельная бд вроде Clickhouse.
❌ Требует сложной настройки и администрирования.
❌ Всё-таки уже устаревает и сейчас тру смузи стек это S3+Iceberg+Trino/Spark

Итог
HDFS + Hive + Spark — это классическая база Big Data-архитектуры. И хотя она уже даже является несколько устаревшей, знание этой базы поможет вам пройти собес и найти работу 💪

Ставьте ❤️ если хотите больше разборов стека Big Data
Ставьте 🔥 если уже сталкивались с этой тройкой в продакшне

@bigdata_postupashki

Товарищи, Поступашкам нужны контент мейкеры. Если вы творческая личность, интересующейся фронтендом, бэкендом, мобильной разработкой, дата сайнс, аналитикой, алгоритмами, вам нравится писать посты/ придумывать идеи для контента, то обязательно пишите @vice22821. Оплата сдельная, ориентировочно за один пост от 2 тыс до 15 тыс рублей.

Обязательно делитесь с ребятами, которым это может быть интересно.

Разбор экзамена по SQL Т-банка

Разбор всех направлений выложен только на наших курсах. До конца дня действует льготная цена 13450 8950. Записываемся!

@ProdAnalysis

задача 1

select count(distinct student_id)
from (
select student_id
from enrollments
where status = 'completed'
) t

задача 2

select s.city
from students s
join enrollments e on s.student_id = e.student_id
where status = 'completed'
group by city
order by sum(e.progress_pct) desc, s.city asc
limit 1;

задача 3

select round(avg(rating)::numeric, 2)
from reviews;

задача 4

with cte_1 as (
select course, student_id, count(student_id) as cnt
from enrollments
where status = 'completed'
group by course, student_id
), cte_2 as (
select course, sum(case when cnt >= 2 then 1 else 0 end) as repeat_students, sum(cnt) as completed
from cte_1
group by course
)
select course
from cte_2
order by repeat_students desc, completed desc, course asc
limit 1;

Ментор+

Эта услуга для тех, кто хочет трудоустроиться уже сейчас без лишней головной боли и теории.

Что мы обещаем
Достигнуть ваши цели, такие как получить оффер на стажировку, получить оффер в Бигтех с грейдом не ниже вашего текущего. Мы обсуждаем сроки. Если мы не в состоянии достигнуть ваших целей, то мы сразу же вам об этом скажем и предложим альтернативу (расширим пул компаний).

Что мы не обещаем
Мы не можем обещать нереальных или почти нереальных результатов как "оффер на 500 тыс (стажер)".
Мы не можем обещать попадание в конкретную компанию.

Как мы достигаем цели?

1. Легенда
Мы прорабатываем с вами легенду, которую вы должны выучить и озвучить на собеседовании: где работали, что делали и так далее. Составляем конкретно под вас резюме.

Если вам нужно проработать легенду и составить резюме, но вы хотите проходить собеседования полностью самостоятельно, то это отдельная услуга. Она стоит 15 тысяч рублей. Для записи: @vice22821.

2. Прохождение собеседований
Подбираем с вами вакансии, проходим с вами собеседования, сидя у вас на наушнике, проходим лайвкодинг с помощью удаленного доступа.

Если вам разово нужен специалист, который поможет на собеседовании: посидит на наушнике или пройдет лайвкодинг с удаленного доступа, то это отдельная услуга. Ориентировочная цена 10 тыс за пол часа собеса. Для записи: @vice22821

3. Теоретические занятия
Теоретические занятия в ментор+ не входят, ментор+ для тех, кому нужно трудоустройство. Если вам нужны знания, вам подойдут курсы.

Если у вас непростая ситуация и требуется консультация специалиста, то это отдельная услуга. Она стоит 8 тыс в час. Для записи: @vice22821.

Если вам нужно мок собеседование, то это отдельная услуга. Она стоит 8 тыс в час. Для записи @vice22821.

Как записаться
Написать @vice22821 со своим запросом. Мы обсудим насколько ваша цель достижима, сколько на это потребуется времени и готовы ли мы взяться за ваш заказ. Если вас все устроит, то начнем работать.

Оплата ментор +
Залог составляет 30 тыс. Уже после получения оффера нужно внести 1 месячную зарплату после вычета налогов.

Что если не получится
Если не получилось получить оффер по нашей вине (после собеседований в 5 компаний у вас нет оффера) в установленные сроки, то мы возвращаем вам залог.

Если не получилось получить оффер по вашей вине (например, вы решили отказаться от сотрудничества) в установленные сроки, то залог не возвращается.

Ответы на часто задаваемые вопросы

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

2. Есть ли сопровождение на испытательном сроке?
Не входит в ментор+, отдельная услуга и обсуждается индивидуально.