Почему spark (sql) не выполняет широковещательное соединение, даже если размер в autoBroadcastJoinThreshold?

Question

Я использую Spark 2.1.1.У меня очень сложный запрос, написанный на Spark SQL, который я пытаюсь оптимизировать.Для раздела я пытаюсь использовать широковещательное соединение.Но хотя я установил:

spark.sql.autoBroadcastJoinThreshold = 1073741824

, что составляет 1 ГБ, я вижу, что сгенерированный искрой физический план для этого раздела выполнениявсе еще использую SortMergeJoin.Есть ли у вас какие-либо сведения о том, почему широковещательное соединение не используется, даже если на одной из сторон Spark UI -> вкладка SQL размер одной стороны показан намного меньше (в МБ)?
Мой раздел кода SQL для уязвимой части выглядит следующим образом:

-- Preceding SQL
(
SELECT  /*+ BROADCAST (a) */   -- Size of a is within broadcast threshold as per UI
     a.id,
     big.gid
 FROM
     (SELECT DISTINCT(id) AS id FROM a_par WHERE gid IS NULL) a
 JOIN
    big ON (a.id=big.id)
)
-- Succeeding SQL

Экран Spark UI, который подтверждает:

Answer 1

Ниже, согласно моим наблюдениям и тому, как я это сделал:

В Spark 2.1:

• В SparkSql:

  • Широковещательный намек НЕ имеет смысла
  • Spark применяет BroadcastHashJoin только в том случае, если он может вычислить размер кадра данных (согласно предыдущему ответу).
  • Это произойдет только в том случае, если одна из сторон соединения является пустым столом (в моем случае это таблица кустов)

• В SparkShell:

  • Мы можем принудительно транслировать кадр данных, используя широковещательную рассылку (df)
  • Если DF не достаточно мал в соответствии с установленными пороговыми значениями, он провалит всю работу.

Короче говоря, я не смог найти способ добиться этого, хотя бы только Spark SQL.Мне пришлось ввести трансляцию (df) для принудительного включения трансляции.

Answer 2

Spark не поддерживает адаптивное выполнение.Он не меняет план выполнения на основе промежуточной статистики (например, размер, макс, мин и т. Д.) После завершения этапа.Таким образом, если план сформирован до выполнения запроса, он не изменяется.Таким образом, вы увидите тот же план.

Причина, по которой spark не передает левую таблицу, заключается в отсутствии статистики вашего подзапроса.Я решил эту проблему путем кеширования результатов вашего запроса.Это помогает зажечь оптимизацию плана.

В вашем случае вы можете сделать что-то вроде:

CACHE TABLE cached_a as SELECT DISTINCT(id) AS id FROM a_par WHERE gid IS NULL;

SELECT
     a.id,
     big.gid
 FROM
      cached_a
 JOIN
    big ON (a.id=big.id)