вот поэтому и спросил, тыкали уже или нет =)

А есть ли теперь смысл в таких тестах функций репартишена?

Может с другой стороны на них посмотреть

Просто учитывать, что мелкие файлы мёрджатся, и проверять что-то другое

ну функция обычно на больших срабатывает объемах (обычно миллионы, ну есть отдельно estimate для avg row size)

просто для тестов там параметр срабатывания поставлен в каких 10-15 записей

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

https://github.com/apache/spark/blob/branch-3.2/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ShufflePartitionsUtil.scala

так понимаю вот это отработало
INFO ShufflePartitionsUtil : For shuffle(0), advisory target size: 67108864, actual target size 2609579, minimum partition size: 1048576

а выше когда планы выбирались

https://github.com/apache/spark/blob/branch-3.2/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala#L88-L98

 // EnsureRequirements may remove user-specified repartition and assume the query plan won't
 // change its output partitioning. This assumption is not true in AQE. Here we check the
 // inputPlan which has not been processed by EnsureRequirements yet, to find out the
 // effective user-specified repartition. Later on, the AQE framework will make sure the final
 // output partitioning is not changed w.r.t the effective user-specified repartition.

то есть "мы умнее пользователя"

Ну вообще сделать адекватный репартишен руками крайне проблематично, учитывая отсутствие понимания количества итоговых рядов без запуска дополнительных запросов

Был бы метод типа row_count_in_partition(), вопросов нет

в общем разобрался

spark.sql.adaptive.coalescePartitions.enabled


val COALESCE_PARTITIONS_ENABLED =
 buildConf("spark.sql.adaptive.coalescePartitions.enabled")
   .doc(s"When true and '${ADAPTIVE_EXECUTION_ENABLED.key}' is true, Spark will coalesce " +
     "contiguous shuffle partitions according to the target size (specified by " +
     s"'${ADVISORY_PARTITION_SIZE_IN_BYTES.key}'), to avoid too many small tasks.")
   .version("3.0.0")
   .booleanConf
 .createWithDefault(true)

по умолчанию оно включено
но для тестов можно бы и отключить

в этом случае оно не пытается мержить мелкие партиции, что идеально для тестов

потому что даже выставление ADVISORY_PARTITION_SIZE_IN_BYTES в малые значения не спасает

так как следом есть другой параметр

val COALESCE_PARTITIONS_PARALLELISM_FIRST =
 buildConf("spark.sql.adaptive.coalescePartitions.parallelismFirst")
   .doc("When true, Spark does not respect the target size specified by " +
     s"'${ADVISORY_PARTITION_SIZE_IN_BYTES.key}' (default 64MB) when coalescing contiguous " +
     "shuffle partitions, but adaptively calculate the target size according to the default " +
     "parallelism of the Spark cluster. The calculated size is usually smaller than the " +
     "configured target size. This is to maximize the parallelism and avoid performance " +
     "regression when enabling adaptive query execution. It's recommended to set this config " +
     "to false and respect the configured target size.")
   .version("3.2.0")
   .booleanConf
 .createWithDefault(true)

в общем спарк всё умнее с каждым релизом
жаль пользователи не умнеют

а какой тогда смысл от такого теста? если в проде точно будет другое поведение.

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

https://github.com/zubtsov/spark-commons/blob/master/src/main/scala/com/epam/bdcc/spark/sql/DataFrameWithId.scala

CC @SparkApplicationMaster

Кстати, если кто хочет помочь с публикацией либы в maven central - пишите в личку)

Спасибо

Суть того, что я скинул - не использовать rdd