Broadcast DataFrame and Filter when joining a big with a small Dataframe

user_3108417 0
Viewed 543

I have a big Dataframe with columns user_id, user_address and multiple user related columns (10-12 more) and a small (< 2MB) Dataframe with columns user_id and user_contact. Each user can have lots of phone numbers/email addresses.

I am trying to broadcast the smaller Dataframe and pass its data to an UDF. However, I am not able to get all the data present in each of the executors. As the UDF returns No match found if no contact details found for that user_id.

I try to filter the broadcastDF by the user_id, but seems like that is not working. Can someone please guide me here?

val DF = createDF("/somePath/")
val broadcastedDF = spark.sparkContext.broadcast(DF)
val DFWithUDF = someDF.select( col("user_id),
                               UDF(col("user_name"),
                               typedLit[Seq[String]](broadcastedDF.value.filter(broadcastedDF.value("user_id") === col("user_id")).
                                                        select("user_contact").
                                                        collect().map(data => 
                                                           String.valueOf(data.getAs[String]("user_contact")))
                                                      )
                                                    ).alias(SCHEMA_REQUEST_TARGET)
apache-spark-sql apache-spark scala
1 Answers

You can use a join hint to broadcast the smaller Dataframe when joining those two Dataframes and there is no need to specifically broadcasting it yourself. Your code could look like:

val bigDf = ... // has columns user_id, user_address
val smallDf = ... // has columns user_id and user_contact

// define your join expression and join those tables using column user_id
import org.apache.spark.sql.functions.broadcast
val joinExpr = bigDf.col("user_id") === smallDf.col("user_id")
val joinedDF = bigDf.join(broadcast(smallDf), joinExpr)

If your small Dataframe is smaller than 10MB and you did not change the configuration spark.sql.autoBroadcastJoinThreshold, which defaults to 10MB you do not even need to mention the broadcast hint in your join.

The configuration spark.sql.autoBroadcastJoinThreshold is described as:

"Configures the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join. By setting this value to -1 broadcasting can be disabled. Note that currently statistics are only supported for Hive Metastore tables where the command ANALYZE TABLE COMPUTE STATISTICS noscan has been run, and file-based data source tables where the statistics are computed directly on the files of data."

user_122089100
Related