Spark SQL broadcast hash join

by

You can explicitly mark the DataFrame as small enough for broadcasting
using broadcast function:

Python:

from pyspark.sql.functions import broadcast

small_df = ...
large_df = ...

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

or broadcast hint (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"))

or broadcast hint (Spark >= 2.2):

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

SQL

You can use hints (Spark >= 2.2):

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 (SparkR):

With hint (Spark >= 2.2):

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

With broadcast (Spark >= 2.3)

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

Note:

Broadcast join is useful if one of structures is relatively small. Otherwise it can be significantly more expensive than a full shuffle.

More Related Contents:

  1. Finding duplicates from large data set using Apache Spark
  2. Using a column value as a parameter to a spark DataFrame function
  3. Find maximum row per group in Spark DataFrame
  4. Split Spark Dataframe string column into multiple columns
  5. While writing to hdfs path getting error java.io.IOException: Failed to rename
  6. Multiple Aggregate operations on the same column of a spark dataframe
  7. Partitioning in spark while reading from RDBMS via JDBC
  8. How to control partition size in Spark SQL
  9. How to import multiple csv files in a single load?
  10. How to access element of a VectorUDT column in a Spark DataFrame?
  11. How to split a list to multiple columns in Pyspark?
  12. How to save/insert each DStream into a permanent table
  13. Use collect_list and collect_set in Spark SQL
  14. Spark load data and add filename as dataframe column
  15. What is the maximum size for a broadcast object in Spark?
  16. PySpark: how to resample frequencies
  17. How does createOrReplaceTempView work in Spark?
  18. PySpark – get row number for each row in a group
  19. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  20. Keep only duplicates from a DataFrame regarding some field
  21. pyspark: count distinct over a window
  22. Using windowing functions in Spark
  23. How can I access python variable in Spark SQL?
  24. PySpark: How to fillna values in dataframe for specific columns?
  25. Why spark application fail with “executor.CoarseGrainedExecutorBackend: Driver Disassociated”?
  26. What does “Correlated scalar subqueries must be Aggregated” mean?
  27. How to improve broadcast Join speed with between condition in Spark
  28. How to calculate Median in spark sqlContext for column of data type double
  29. Spark SQL referencing attributes of UDT
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment