// Use ShuffledHashJoinExec's selection requirements
// 1. Disable auto broadcasting
// JoinSelection (canBuildLocalHashMap specifically) requires that
// plan.stats.sizeInBytes < autoBroadcastJoinThreshold * numShufflePartitions
// That gives that autoBroadcastJoinThreshold has to be at least 1
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 1)

да, броадкаст join тоже отключал

spark.conf.set("spark.sql.join.preferSortMergeJoin", false)

Хинт поставьте.

df2. hint("shuffle_hash").join( df1,"id")

Тоже не всегда помогает

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

спасибо, попробую

вот так можно:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 1)
spark.conf.set("spark.sql.join.preferSortMergeJoin", false)
spark.conf.set("spark.sql.adaptive.enabled", false)

val dataset = Seq(
  (0, "playing"),
  (1, "with"),
  (2, "ShuffledHashJoinExec")
).toDF("id", "token")

val q = dataset.hint("shuffle_hash").join(dataset, Seq("id"))
q.explain

ответ:

== Physical Plan ==
*(1) Project [id#231, token#232, token#236]
+- ShuffledHashJoin [id#231], [id#235], Inner, BuildLeft, false
   :- Exchange hashpartitioning(id#231, 200), true, [id=#406]
   :  +- LocalTableScan [id#231, token#232]
   +- ReusedExchange [id#235, token#236], Exchange hashpartitioning(id#231, 200), true, [id=#406]


dataset: org.apache.spark.sql.DataFrame = [id: int, token: string]
q: org.apache.spark.sql.DataFrame = [id: int, token: string ... 1 more field]

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 1)
spark.conf.set("spark.sql.join.preferSortMergeJoin", false)
spark.conf.set("spark.sql.adaptive.enabled", false)

val dataset = Seq(
  (0, "playing"),
  (1, "with"),
  (2, "ShuffledHashJoinExec")
).toDF("id", "token")

val q = dataset.hint("shuffle_hash").join(dataset, Seq("id"))
q.explain

спасибо, в датабриксе еще сработало с хинтом в sql

/*+ SHUFFLE_HASH(m, i) */

Добрый день.
С информатикой 10.4.01 работал  кто? Сталкивались ли вы с проблемой постановки маппингов с статус queue?

@renardeinside , можно гдето почитать про внутреннее устройство z-order? что он делает под капотом? Сортирует, бакетирует, создает ли новый файлы статистик, дополняет ли parquet файлы новыми данными?

https://docs.databricks.com/delta/optimizations/file-mgmt.html#z-ordering-multi-dimensional-clustering
https://databricks.com/blog/2018/07/31/processing-petabytes-of-data-in-seconds-with-databricks-delta.html?_ga=2.39107680.2124548659.1604511640-1543332958.1604511640

вторая больше техническая, там и Zorder объясняется

Да работаем.

Коллеги, есть вопрос. Имеется керберизированный кластер hadoop, с настроенным KNOX (Yarn, HDFS, HIVE). Есть необходимость запускать задачи PySPARK через KNOX. Подскажите куда копать для настройки spark-submit, для запуска через KNOX.

В livy

Livy наружу не торчит.

Можно выставить

Knox умеет в livy ходить

Можно, но долго ... наш knox официально не может ... бюрократия мешает.