How to access broadcasted DataFrame in Spark

Question

I have created two dataframes which are from Hive tables(PC_ITM and ITEM_SELL) and big in size and I am using those frequently in the SQL query by registering as table.But as those are big, it is taking much time to get the query result.So I have saved them as parquet file and then read them and registered as temporary table.But still I am not getting good performance so I have broadcasted those data-frames and then registered as tables as below.

PC_ITM_DF=sqlContext.parquetFile("path")
val PC_ITM_BC=sc.broadcast(PC_ITM_DF)
val PC_ITM_DF1=PC_ITM_BC
PC_ITM_DF1.registerAsTempTable("PC_ITM")

ITM_SELL_DF=sqlContext.parquetFile("path")
val ITM_SELL_BC=sc.broadcast(ITM_SELL_DF)
val ITM_SELL_DF1=ITM_SELL_BC.value
ITM_SELL_DF1.registerAsTempTable(ITM_SELL)


sqlContext.sql("JOIN Query").show

But still I cant achieve performance it is taking same time as when those data frames are not broadcasted.

Can anyone tell if this is the right approach of broadcasting and using it?`

Answer 1

You don't really need to 'access' the broadcast dataframe - you just use it, and Spark will implement the broadcast under the hood. The broadcast function works nicely, and makes more sense that the sc.broadcast approach.

It can be hard to understand where the time is being spent if you evaluate everything at once.

You can break your code into steps. The key here will be performing an action and persisting the dataframes you want to broadcast before you use them in your join.

// load your dataframe
PC_ITM_DF=sqlContext.parquetFile("path")

// mark this dataframe to be stored in memory once evaluated
PC_ITM_DF.persist()

// mark this dataframe to be broadcast
broadcast(PC_ITM_DF)

// perform an action to force the evaluation
PC_ITM_DF.count()

Doing this will ensure that the dataframe is

• loaded in memory (persist)
• registered as temp table for use in your SQL query
• marked as broadcast, so will be shipped to all executors

When you now run sqlContext.sql("JOIN Query").show you should now see a 'broadcast hash join' in the SQL tab of your Spark UI.