Может там захинтованный запрос или может Спарк по какой причине неправильно размер вычисляет. Как можно быть уверенным , что он броадкастит правильно ,если код пишут люди ;)

он пишет что на одном кластере это работает, на другом нет, 3 тб бродкаста это на каком кластере влезет?

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

Бывает такое иногда - есть гигантская таблица и таблица на 100 записей, пишешь броадкаст мелкой таблицы, а он броадкастит большую вместо мелкой

И ничего не работает

Приходится только либо резать где-то план запроса (cache(), например), либо отключать броадкаст

Ребята, привет!
Подскажите, пожалуйста, можно ли сделать методом RDD.zipPartitions вот такое:

1. Если принять, что к rdd1 и rdd2 уже применен один и тот же партишенер и выполнен .sortWithinPartitions,
можно ли на этапе .zipPartitions сделать что-то вроде mergeSort, объединить два отсортированных итератора в один отсортированный?
Вот так очень просто: rdd1.zipPartitions(rdd2) { (iter1, iter2) => iter1 ++ iter2 }
но совсем не хочется после сортировки и слияния делать еще одну сортировку

2. Произвести anti join, то есть, сделать примерно так:
def getKey(row: Row) = ???
rdd1.zipPartitions(rdd2) {
 (iter1, iter2) => iter1.filterNot( row => iter2map(getKey).toSet.contains(getKey(row)))
}

Буду очень благодарен за конкретные примеры

В инженерский чат закинул этот же вопрос

в любом случае если массив уже отсортированный, его сортировка должна быть крайне быстра

Всем привет, может кто сталкивался, использую Structured Streaming (2.3.0) - метрики (InfluxDbSink) не публикуются куда надо.

Конфигурировании пробовал разные, в том числе вот так:

SparkSession
.builder()
.config("spark.sql.streaming.metricsEnabled",value = true)
.config("spark.metrics.namespace","new_streamer")

Конфигурация Influx есть в кластере и стандартные метрики (batch records, scheduling delay and etc.) успешно отгружаются если применять RDD подход, но с SS никак не могу получить их 🙁

Подскажите куда копать?

SparkSession
.builder()
.config("spark.sql.streaming.metricsEnabled",value = true)
.config("spark.metrics.namespace","new_streamer")

А попробуйте value="true" (строчкой), вдруг сработает

Сейас проверю

к сожалению не помогло 🙁

может есть какой-то способ проверить что эти метрики собриаются и им проставляется корректный namespace ?

Всем привет! Может кто-то оперативно подсказать, как в pyspark при записи датафрейма пробрасывать в метаинформацию какую-нибудь строку и как потом её считывать? Что-то вроде description/комментария у колонок, только у всей таблицы.

Нашёл. Делается через:

Может, кому-то тоже будет полезно :)

dic... Такого сокращения я не видел еще😂

на leetcode так часто пишут