-->

Problem

PySpark Join is used to combine two DataFrames and by chaining these you can join multiple DataFrames; it supports all basic join type operations available in traditional SQL like INNER, LEFT OUTER, RIGHT OUTER, LEFT ANTI, LEFT SEMI, CROSS, SELF JOIN.PySpark Joins are wider transformations that involve data shuffling across the network.
DataFrames tutorial. The Apache Spark DataFrame API provides a rich set of functions (select columns, filter, join, aggregate, and so on) that allow you to solve common data analysis problems efficiently. DataFrames also allow you to intermix operations seamlessly with custom Python, SQL, R, and Scala code. This tutorial module shows how to.
SparkHub is the community site of Apache Spark, providing the latest on spark packages, spark releases, news, meetups, resources and events all in one place.

The easiest way to start working with DataFrames is to use an example Databricks dataset available in the /databricks-datasets folder accessible within the Databricks workspace. To access the file that compares city population versus median sale prices of homes, load the file /databricks-datasets/samples/population-vs-price/datageo.csv.

You are attempting to join two large tables, projecting selected columns from the first table and all columns from the second table.

Despite the total size exceeding the limit set by spark.sql.autoBroadcastJoinThreshold, BroadcastHashJoin is used and Apache Spark returns an OutOfMemorySparkException error.

Cause

Databricks Python Join Two Dataframes

This is due to a limitation with Spark’s size estimator.

If the estimated size of one of the DataFrames is less than the autoBroadcastJoinThreshold, Spark may use BroadcastHashJoin to perform the join. If the available nodes do not have enough resources to accommodate the broadcast DataFrame, your job fails due to an out of memory error.

Solution

There are three different ways to mitigate this issue.

Use ANALYZE TABLE to collect details and compute statistics about the DataFrames before attempting a join.
Cache the table you are broadcasting.
1. Run explain on your join command to return the physical plan.
2. Review the physical plan. If the broadcast join returns BuildLeft, cache the left side table. If the broadcast join returns BuildRight, cache the right side table.
In Databricks Runtime 7.0 and above, set the join type to SortMergeJoin with join hints enabled.

When you perform a join command with DataFrame or Dataset objects, if you find that the query is stuck on finishing a small number of tasks due to data skew, you can specify the skew hint with the hint('skew') method: df.hint('skew'). The skew join optimization is performed on the DataFrame for which you specify the skew hint.

In addition to the basic hint, you can specify the hint method with the following combinations of parameters: column name, list of column names, and column name and skew value.

DataFrame and column name. The skew join optimization is performed on the specified column of the DataFrame.
DataFrame and multiple columns. The skew join optimization is performed for multiple columns in the DataFrame.
DataFrame, column name, and skew value. The skew join optimization is performed on the data in the column with the skew value.

Example

Join Two Dataframes Spark

This example shows how to specify the skew hint for multiple DataFrame objects involved in a join operation:

Databricks Join Dataframes