星火SQL广播哈希联接星火SQL广播哈希联接(Spark SQL broadcast hash jo

我试图使用SparkSQL dataframes执行广播哈希联接如下记载： https://docs.cloud.databricks.com/docs/latest/databricks_guide/06%20Spark%20SQL%20%26%20DataFrames/05% 20BroadcastHashJoin％20-％20scala.html

在这个例子中，（小） DataFrame通过saveAsTable依然存在，再有就是通过火花SQL联接（即通过sqlContext.sql("..."))

我的问题是，我需要使用sparkSQL API来构建我的SQL（我左侧的接合部〜50桌与ID列表，并且不希望手工编写的SQL）。

How do I tell spark to use the broadcast hash join via the API?  The issue is that if I load the ID list (from the table persisted via `saveAsTable`) into a `DataFrame` to use in the join, it isn't clear to me if Spark can apply the broadcast hash join.

Answer 1:

你可以明确地标记DataFrame的足够小，使用广播broadcast功能：

Python的 ：

from pyspark.sql.functions import broadcast

small_df = ...
large_df = ...

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

或广播的提示（火花> = 2.2）：

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

斯卡拉 ：

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

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

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

或广播的提示（火花> = 2.2）：

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

SQL

您可以使用提示（星火> = 2.2 ）：

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

要么

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

要么

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

R（SparkR）：

与hint （火花> = 2.2）：

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

与broadcast （火花> = 2.3）

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

注意：

广播加入是有用的，如果结构中的一个相对较小。否则，它可以显著不是一个完整的洗牌更加昂贵。

Answer 2:

jon_rdd = sqlContext.sql( "select * from people_in_india  p
                            join states s
                            on p.state = s.name")


jon_rdd.toDebugString() / join_rdd.explain() :

shuffledHashJoin：
所有印度的数据将被洗牌成只有29对每个国家的钥匙。问题：分片不均匀。有限的并行与29个的输出分区。

broadcaseHashJoin：

广播小RDD为所有的工作节点。大RDD的并行性依然维持和洗牌甚至不要求。