DataFrame join optimization – Broadcast Hash Join

by

Broadcast Hash Joins (similar to map side join or map-side combine in Mapreduce) :

In SparkSQL you can see the type of join being performed by calling queryExecution.executedPlan. As with core Spark, if one of the tables is much smaller than the other you may want a broadcast hash join. You can hint to Spark SQL that a given DF should be broadcast for join by calling method broadcast on the DataFrame before joining it

Example:
largedataframe.join(broadcast(smalldataframe), "key")

in DWH terms, where largedataframe may be like fact
smalldataframe may be like dimension

As described by my fav book (HPS) pls. see below to have better understanding..
enter image description here

Note : Above broadcast is from import org.apache.spark.sql.functions.broadcast not from SparkContext

Spark also, automatically uses the spark.sql.conf.autoBroadcastJoinThreshold to determine if a table should be broadcast.

Tip : see DataFrame.explain() method

def
explain(): Unit
Prints the physical plan to the console for debugging purposes.

Is there a way to force broadcast ignoring this variable?

sqlContext.sql("SET spark.sql.autoBroadcastJoinThreshold = -1")

NOTE :

Another similar out of box note w.r.t. Hive (not spark) : Similar
thing can be achieved using hive hint MAPJOIN like below…

Select /*+ MAPJOIN(b) */ a.key, a.value from a join b on a.key = b.key

hive> set hive.auto.convert.join=true;
hive> set hive.auto.convert.join.noconditionaltask.size=20971520
hive> set hive.auto.convert.join.noconditionaltask=true;
hive> set hive.auto.convert.join.use.nonstaged=true;
hive> set hive.mapjoin.smalltable.filesize = 30000000; // default 25 mb made it as 30mb

Further Reading : Please refer my article on BHJ, SHJ, SMJ

More Related Contents:

  1. Spark SQL: apply aggregate functions to a list of columns
  2. Multiple Aggregate operations on the same column of a spark dataframe
  3. How to access element of a VectorUDT column in a Spark DataFrame?
  4. What is the maximum size for a broadcast object in Spark?
  5. How to loop through each row of dataFrame in pyspark
  6. Why does Spark think this is a cross / Cartesian join
  7. How do I add an persistent column of row ids to Spark DataFrame?
  8. Convert null values to empty array in Spark DataFrame
  9. How to exclude multiple columns in Spark dataframe in Python
  10. Why is predicate pushdown not used in typed Dataset API (vs untyped DataFrame API)?
  11. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  12. Find maximum row per group in Spark DataFrame
  13. While writing to hdfs path getting error java.io.IOException: Failed to rename
  14. Spark Dataframe distinguish columns with duplicated name
  15. How to import multiple csv files in a single load?
  16. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  17. Requirements for converting Spark dataframe to Pandas/R dataframe
  18. Retrieve top n in each group of a DataFrame in pyspark
  19. Provide schema while reading csv file as a dataframe
  20. Create Spark DataFrame. Can not infer schema for type
  21. Renaming column names of a DataFrame in Spark Scala
  22. What is going wrong with `unionAll` of Spark `DataFrame`?
  23. spark dataframe drop duplicates and keep first
  24. pyspark: count distinct over a window
  25. How to get keys and values from MapType column in SparkSQL DataFrame
  26. reduce result datasets into single dataset
  27. Filtering DataFrame using the length of a column
  28. Stratified sampling with pyspark
  29. What is an optimized way of joining large tables in Spark SQL
  30. Difference between === null and isNull in Spark DataDrame

Leave a Comment