Spark Shared Variables

In Apache Spark, shared variables enable multiple tasks to share data efficiently on each worker node while processing a job. Spark has two types of shared variables: Broadcast variables and Accumulators.

Broadcast Variables

In Spark RDD and DataFrame, Broadcast variables are read-only shared variables that are cached and available on all nodes in a cluster to be accessed or used by the tasks. Instead of sending this data along with every task, Spark distributes broadcast variables to the machine using efficient broadcast algorithms to reduce communication costs.

Use case

Let's assume you are getting a two-letter country state code in a file and you want to transform it to a full state name (for example, CA to California, NY to New York, etc.) by doing a lookup to reference mapping. In some instances, this data could be large, and you may have many such lookups (like zip code).

Instead of distributing this information along with each task over the network (which is overhead and time-consuming), we can use the broadcast variable to cache this lookup info on each machine, and tasks use this cached info while executing the transformations.

How does Spark Broadcast work?

Broadcast variables are used in the same way for RDD, DataFrame, and Dataset. In this lesson, I will primarily focus on RDD.

When you run a Spark RDD job that has Broadcast variables defined and used, Spark does the following:

• Spark breaks the job into stages that have distributed shuffling, and actions are executed within the stage.
• Later stages are also broken into tasks.
• Spark broadcasts the common data (reusable) needed by tasks within each stage.
• The broadcasted data is cached in serialized format and deserialized before executing each task.

You should be creating and using broadcast variables for data shared across multiple stages and tasks.

Creating Broadcast variable

The Spark Broadcast is created using the broadcast(v) method of the SparkContext class. This method takes the argument v that you want to broadcast.

In Java

JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("example").setMaster("local"));
Broadcast<int[]> broadcastVar = sc.broadcast(new int[]{0, 1, 2, 3});
int[] values = broadcastVar.value();

Note that broadcast variables are not sent to executors when you call sc.broadcast(variable) statement but are sent to executors when they are first used.

Spark RDD Broadcast Variable Example

Below is a very simple example of how to use broadcast variables on RDD. This example defines commonly used data (country and states) in a Map variable and distributes the variable using SparkContext.broadcast() and then uses these variables on an RDD map() transformation.

// Imports
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.SparkConf;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class RDDBroadcast {

  public static void main(String[] args) {

    // Create SparkContext
    SparkConf conf = new SparkConf().setAppName("AIStudyLabs.com").setMaster("local");
    JavaSparkContext sc = new JavaSparkContext(conf);

    Map<String, String> states = new HashMap<>();
    states.put("NY", "New York");
    states.put("CA", "California");
    states.put("FL", "Florida");

    Map<String, String> countries = new HashMap<>();
    countries.put("USA", "United States of America");
    countries.put("IN", "India");

    // Create broadcast variable
    final Broadcast<Map<String, String>> broadcastStates = sc.broadcast(states);
    final Broadcast<Map<String, String>> broadcastCountries = sc.broadcast(countries);

    // Create RDD
    List<String[]> data = Arrays.asList(
      new String[]{"James", "Smith", "USA", "CA"},
      new String[]{"Michael", "Rose", "USA", "NY"},
      new String[]{"Robert", "Williams", "USA", "CA"},
      new String[]{"Maria", "Jones", "USA", "FL"}
    );

    JavaRDD<String[]> rdd = sc.parallelize(data);

    // RDD Transformation
    JavaRDD<String[]> rdd2 = rdd.map(f -> {
      String country = f[2];
      String state = f[3];

      // Using broadcast variables
      String fullCountry = broadcastCountries.value().get(country);
      String fullState = broadcastStates.value().get(state);
      return new String[]{f[0], f[1], fullCountry, fullState};
    });

    // Print RDD
    rdd2.collect().forEach(arr -> System.out.println(String.join(",", arr)));
    sc.close();
  }
}