"Возможен ли Sort-Merge Join без шафла? Ага, в спарке есть такая оптимизация - Storage Partition Join (SPJ). Если таблицы одинаково партиционированы и джойнятся по этому ключу, то не приходится перемещать кучу данных. Правда, это работает только в форматах, поддерживающих DataSource API V2. Например, в Iceberg, Hudi, Delta Lake Что нужно: 🌸 Spark 3.3+ 🌸 Условие равенства в джойне 🌸 У таблиц должно быть общее поле-партиция, которое входит в ключ джойна 🎈 Пример Я взяла 2 таблицы, партицировала, заполнила данными и положила. Потом их поджойнила, и вот какой план запроса у меня сначала получился (картинка 1): ``` == Physical Plan == CollectLimit +- * Project +- * SortMergeJoin Inner :- * Sort : +- Exchange : +- * Project : +- BatchScan spark_catalog.test.table1 +- * Sort +- Exchange +- * Project +- BatchScan spark_catalog.test.table2 ``` У нас есть Exchange - происходит шафл даных. Потом я накинула несколько конфигов: ``` # включает Storage Partition Join spark.conf.set(""spark.sql.sources.v2.bucketing.enabled"", ""true"") # сохраняет текущее партиционирование при планировании запроса spark.conf.set(""spark.sql.iceberg.planning.preserve-data-grouping"", ""true"") # убирает шафл, когда партиции между таблицами не совпадают по количеству spark.conf.set(""spark.sql.sources.v2.bucketing.pushPartValues.enabled"", ""true"") # иначе ключи джойна должны быть такие же, как партиции, в том же порядке spark.conf.set(""spark.sql.requireAllClusterKeysForCoPartition"", ""false"") # борется с перекосом данных spark.conf.set(""spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled"", ""true"") # разрешает включать части партиций в ключ джойна spark.conf.set(""spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled"", ""true"") # на всякий можно отключить broadcast, адаптивку и sortmerge (если это точно нужно) spark.conf.set(""spark.sql.autoBroadcastJoinThreshold"", -1) spark.conf.set(""spark.sql.adaptive.enabled"", ""false"") spark.conf.set(""spark.sql.join.preferSortMergeJoin"", ""false"") ``` План запроса чуть поменялся (картинка 2) - исчез шафл: ``` == Physical Plan == CollectLimit +- * Project +- * SortMergeJoin Inner :- * Sort : +- * Project : +- BatchScan spark_catalog.test.table1 +- * Sort +- * Project +- BatchScan spark_catalog.test.table2``` В одной из статей говорится, что перформанс увеличился на 45-70% на джойнах. А кто-то уже использовал на практике? @data_engineerette"