GP
экспрешоны
Size: a a a
2021 October 26
ИК
rdd и df связаны, так что не просто кусок плана
ИК
шафл структуры, например, вполне материальны, так что от памяти полностью абстрагироваться не выйдет
A
кстати раз зашла речь про персист
df2 = df.persis()
df2.isEmpty() - считает один раз и не вызывает реально кеширование датафрейма
df2.count() - вот тут уже посчитает весь датафрейм и заперсистит его
я понимаю что isEmpty имеет упрощённый план до первой непустой партиции
но на persist датафрейме я бы ожидал что оно тригернёт вычисление и кеширование указанного дф чтобы потом след операцией я уже работал с кешированным
а так получается первый раз isEmpty молотит
потом второй раз молотит когда я дествительно с фреймом работаю
может кто-то объяснить логику или я хочу слишком много?
df2 = df.persis()
df2.isEmpty() - считает один раз и не вызывает реально кеширование датафрейма
df2.count() - вот тут уже посчитает весь датафрейм и заперсистит его
я понимаю что isEmpty имеет упрощённый план до первой непустой партиции
но на persist датафрейме я бы ожидал что оно тригернёт вычисление и кеширование указанного дф чтобы потом след операцией я уже работал с кешированным
а так получается первый раз isEmpty молотит
потом второй раз молотит когда я дествительно с фреймом работаю
может кто-то объяснить логику или я хочу слишком много?
ИК
с ходу ставлю на полноценный персист при вызове .isEmpty
ИК
но надо глянуть))
A
вот посмотри, там как раз и нету его
A
я первый раз когда столкнулся полез смотреть планы
A
в самом спарке там где нужно заперсистить гарантированно а не lazy везде стоит .count()
ИК
у меня только Option.isEmpty)) щас раскопаю
A
/**
* @note Due to complications in the internal implementation, this method will raise an
* exception if called on an RDD of
* because, for example, the type of
* (`parallelize(Seq())` should be avoided anyway in favor of
* @return true if and only if the RDD contains no elements at all. Note that an RDD
* may be empty even when it has at least 1 partition.
*/
def isEmpty(): Boolean = withScope {
partitions.length == 0 || take(1).length == 0
}
https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L1432-L1443
* @note Due to complications in the internal implementation, this method will raise an
* exception if called on an RDD of
Nothing or Null. This may be come up in practice* because, for example, the type of
parallelize(Seq()) is RDD[Nothing].* (`parallelize(Seq())` should be avoided anyway in favor of
parallelize(Seq[T]()).)* @return true if and only if the RDD contains no elements at all. Note that an RDD
* may be empty even when it has at least 1 partition.
*/
def isEmpty(): Boolean = withScope {
partitions.length == 0 || take(1).length == 0
}
https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/RDD.scala#L1432-L1443
A
@KaiNie_R то есть если повезёт то все партиции посчитаются, если не повезет то только некоторые (до первой ненулевой, ну и после неё ещё немного, в зависимости от батча)
ИК
это RDD так проверяется, в датасете чуть запутаннее
A
примерно то на то
датасет проваливается в эту функцию
датасет проваливается в эту функцию
ИК
у меня была одна проверка df.rdd.isEmpty, и я её выполнял на rdd после join и groupBy, по итогу убрал вообще, надоело ждать
A
там такая же логика на numPartsToTry и потом по этим партициям скан идёт батчем
не нашли давайте следующие партиции сравнивать
не нашли давайте следующие партиции сравнивать
A
можешь по ссылкам код сравнить =)
A
я просто тоже много где встречал что люди тригерят кеш через isEmpty
ИК
надо рассказать, что код дублировать не бест практис))