Spark SQL 广播哈希连接

我正在尝试使用 SparkSQL 对数据帧执行广播哈希连接记录在这里 https://spark.apache.org/docs/latest/sql-performance-tuning.html#join-strategy-hints-for-sql-queries.

在该示例中，（小）DataFrame通过 saveAsTable 持久化，然后通过 Spark SQL 进行连接（即通过sqlContext.sql("..."))

我遇到的问题是我需要使用 SparkSQL API 来构建我的 SQL（我需要用 ID 列表连接约 50 个表，并且不想手动编写 SQL）。

如何通过 API 告诉 Spark 使用广播哈希连接？问题是，如果我加载 ID 列表（通过持久化的表）saveAsTable) 变成DataFrame要在连接中使用，我不清楚 Spark 是否可以应用广播哈希连接。

您可以明确标记DataFrame小到足以进行广播 使用broadcast功能：

Python:

from pyspark.sql.functions import broadcast

small_df = ...
large_df = ...

large_df.join(broadcast(small_df), ["foo"])

或广播提示（Spark >= 2.2）：

large_df.join(small_df.hint("broadcast"), ["foo"])

Scala:

import org.apache.spark.sql.functions.broadcast

val smallDF: DataFrame = ???
val largeDF: DataFrame = ???

largeDF.join(broadcast(smallDF), Seq("foo"))

或广播提示（Spark >= 2.2）：

largeDF.join(smallDF.hint("broadcast"), Seq("foo"))

SQL

您可以使用提示（火花 >= 2.2 https://issues.apache.org/jira/browse/SPARK-16475):

SELECT /*+ MAPJOIN(small) */ * 
FROM large JOIN small
ON large.foo = small.foo

or

SELECT /*+  BROADCASTJOIN(small) */ * 
FROM large JOIN small
ON large.foo = small.foo

or

SELECT /*+ BROADCAST(small) */ * 
FROM large JOIN small
ON larger.foo = small.foo

R（火花R）：

With hint（火花 >= 2.2）：

join(large, hint(small, "broadcast"), large$foo == small$foo)

With broadcast（火花 >= 2.3）

join(large, broadcast(small), large$foo == small$foo)

Note:

如果其中一个结构相对较小，则广播连接很有用。否则它可能比完全洗牌要贵得多。