Всем привет! Возник вопрос по оптимизации. Как можно повлиять на план запроса, который строит Spark, а именно указать правильный порядок джойна? Знаю, что есть хинты для broadcast join, но у меня ситуация другая. Джойнятся две большие таблицы факта (условно F1 и F2) + каждая из них джойнится на календарь для фильтрации (D). Spark строит план ((F1 broadcast join D) sortmerge join F2) broadcast join D). Вместо того, что вторую таблицу фактов так же предварительно фильтрануть по календарю. Сталкивался кто с таким?

Всем привет! Возник вопрос по оптимизации. Как можно повлиять на план запроса, который строит Spark, а именно указать правильный порядок джойна? Знаю, что есть хинты для broadcast join, но у меня ситуация другая. Джойнятся две большие таблицы факта (условно F1 и F2) + каждая из них джойнится на календарь для фильтрации (D). Spark строит план ((F1 broadcast join D) sortmerge join F2) broadcast join D). Вместо того, что вторую таблицу фактов так же предварительно фильтрануть по календарю. Сталкивался кто с таким?

Можно попробовать посмотреть на кастомные оптимизации https://www.waitingforcode.com/apache-spark-sql/introduction-custom-optimization-apache-spark-sql/read

Cbo не для всех случаев работает.

Господа, всем привет. Мне тут достался легаси spark-streaming 1.6 джоб, в нем несколько оригинальных решений.
Оригинальное решение 1: джоб процессит довольно длинный список топиков кафки, на каждый топик эксплицитно создается  new Thread на драйвере (в который передается ssc).
Оригинальное решение номер 2: там написана хитрая обвязка, которая хранит оффсеты в Hbase! Целевая кафка была 0.9
В таком вообще был какой-то смысл даже во времена 1.6?

Господа, всем привет. Мне тут достался легаси spark-streaming 1.6 джоб, в нем несколько оригинальных решений.
Оригинальное решение 1: джоб процессит довольно длинный список топиков кафки, на каждый топик эксплицитно создается  new Thread на драйвере (в который передается ssc).
Оригинальное решение номер 2: там написана хитрая обвязка, которая хранит оффсеты в Hbase! Целевая кафка была 0.9
В таком вообще был какой-то смысл даже во времена 1.6?

хранить оффсеты имело смысл да и имеет; когда руками коммит хочется сделать и рекавери сделать с руками записаных оффсетов

А джоба в режиме fair scheduler запускается? Иначе от тред пула, как бэ нет толку.

угу, в fair

а вот это разве не коммит "руками"?
(пример из спарковой доки)
stream.foreachRDD { rdd =>
 val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
 // some time later, after outputs have completed
 stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}
в чем вообще смысл хранить оффсет "снаружи" (при том, что хранится не история, а один последний оффсет)

Чекпоинты по умолчанию нужны для stateful streaming, просто проблемка там в том, что при изменении кодовой базы они невалидны становятся