What are broadcast variables? What problems do they solve?

by

If you have a huge array that is accessed from Spark Closures, for example, some reference data, this array will be shipped to each spark node with closure. For example, if you have 10 nodes cluster with 100 partitions (10 partitions per node), this Array will be distributed at least 100 times (10 times to each node).

If you use broadcast, it will be distributed once per node using an efficient p2p protocol.

val array: Array[Int] = ??? // some huge array
val broadcasted = sc.broadcast(array)

And some RDD

val rdd: RDD[Int] = ???

In this case, array will be shipped with closure each time

rdd.map(i => array.contains(i))

and with broadcast, you’ll get a huge performance benefit

rdd.map(i => broadcasted.value.contains(i))

Broadcast variables are used to send shared data (for example application configuration) across all nodes/executors.

The broadcast value will be cached in all the executors.

Sample scala code creating broadcast variable at driver:

val broadcastedConfig:Broadcast[Option[Config]] = sparkSession.sparkContext.broadcast(objectToBroadcast)

Sample scala code receiving broadcasted variable at executor side:

val config =  broadcastedConfig.value

More Related Contents:

  1. How to connect Spark SQL to remote Hive metastore (via thrift protocol) with no hive-site.xml?
  2. Spark SQL – load data with JDBC using SQL statement, not table name
  3. Spark Dataframe validating column names for parquet writes
  4. How to optimize partitioning when migrating data from JDBC source?
  5. Spark: subtract two DataFrames
  6. How to improve performance for slow Spark jobs using DataFrame and JDBC connection?
  7. TypeError: Column is not iterable – How to iterate over ArrayType()?
  8. How to overwrite the output directory in spark
  9. Is groupByKey ever preferred over reduceByKey
  10. Filtering a spark dataframe based on date
  11. Pyspark : forward fill with last observation for a DataFrame
  12. spark.sql.crossJoin.enabled for Spark 2.x
  13. Spark DataFrame Schema Nullable Fields
  14. Why do I have to explicitly tell Spark what to cache?
  15. multiple conditions for filter in spark data frames
  16. Why does Spark think this is a cross / Cartesian join
  17. Why does sortBy transformation trigger a Spark job?
  18. How to pass whole Row to UDF – Spark DataFrame filter
  19. Spark SQL broadcast hash join
  20. What is the difference between cache and persist?
  21. Rename more than one column using withColumnRenamed
  22. What is the difference between Apache Spark SQLContext vs HiveContext?
  23. Saving dataframe to local file system results in empty results
  24. Dealing with a large gzipped file in Spark
  25. What should be the optimal value for spark.sql.shuffle.partitions or how do we increase partitions when using Spark SQL?
  26. pyspark: Efficiently have partitionBy write to same number of total partitions as original table
  27. Spark lists all leaf node even in partitioned data
  28. Spark gives a StackOverflowError when training using ALS
  29. AWS EMR – ModuleNotFoundError: No module named ‘pyarrow’
  30. PySpark: How to fillna values in dataframe for specific columns?

Leave a Comment