Introduction

In Spark, foreachPartition() is used when you have a heavy initialization (like a database connection) and want to initialize once per partition, whereas foreach() is used to apply a function on every element of an RDD/DataFrame/Dataset partition.

Spark foreachPartition is an action operation available in RDD, DataFrame, and Dataset. This is different from other actions as foreachPartition() function doesn’t return a value; instead, it executes input functions on each partition.

1. DataFrame foreachPartition() Usage

On Spark DataFrame, foreachPartition() is similar to the foreach() action used to manipulate accumulators, write to a database table, or external data sources. The difference is that foreachPartition() gives you an option to do heavy initializations per each partition and is considered most efficient.

When foreachPartition() is applied on Spark DataFrame, it executes a function specified in foreach() for each partition on the DataFrame. This operation is mainly used if you want to save the DataFrame result to RDBMS tables or produce it to Kafka topics, etc.

Example

In this example, to make it simple, we just print the DataFrame to the console.

// ForeachPartition DataFrame
List<Row> data = Arrays.asList(
  RowFactory.create("Product1", 100, "USA"),
  RowFactory.create("Product2", 200, "UK")
);

StructType schema = new StructType(new StructField[]{
  new StructField("Product", DataTypes.StringType, false, Metadata.empty()),
  new StructField("Amount", DataTypes.IntegerType, false, Metadata.empty()),
  new StructField("Country", DataTypes.StringType, false, Metadata.empty())
});

Dataset<Row> df = spark.createDataFrame(data, schema);

df.foreachPartition(partition -> {
  // Initialize database connection or Kafka
  while (partition.hasNext()) {
    Row row = partition.next();
    // Apply the function to insert into the database
    // Or produce Kafka topic
    System.out.println(row);
  }
  // If you have batch inserts, do here.
});

// Sample printed data output
Product1  100  USA
Product2  200  UK
...

2. DataFrame foreach() Usage

When foreach() is applied on Spark DataFrame, it executes a function specified for each element of DataFrame/Dataset. This operation is mainly used if you want to manipulate accumulators and any other operations which don’t have heavy initializations. You will learn more about accumulators in the next chapters.

// DataFrame foreach() Usage
LongAccumulator longAcc = spark.sparkContext().longAccumulator("SumAccumulator");

df.foreach(f -> {
  longAcc.add(f.getInt(1));
});
System.out.println("Accumulator value: " + longAcc.value());

Accumulator value: 300