Introduction

In PySpark, 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.

DataFrame foreach() Usage

When foreach() is applied on a PySpark 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
from pyspark import SparkContext

sc = SparkContext.getOrCreate()
long_acc = sc.longAccumulator("SumAccumulator")

def process_row(row):
    long_acc.add(row['Amount'])

df.foreach(process_row)
print("Accumulator value:", long_acc.value)

Accumulator value: 300

DataFrame foreachPartition() Usage

The foreachPartition() is similar to the foreach() action used to manipulate the 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 a PySpark 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
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("example").getOrCreate()
data = [("Product1", 100, "USA"), ("Product2", 200, "UK")]
df = spark.createDataFrame(data, ["Product", "Amount", "Country"])

def process_partition(partition):
    # Initialize database connection or Kafka
    for row in partition:
        # Apply the function to insert into the database
        # Or produce to Kafka topic
        print(row)

df.foreachPartition(process_partition)

# Sample printed data output
Row(Product='Product1', Amount=100, Country='USA')
Row(Product='Product2', Amount=200, Country='UK')
...