The foreach() function in Spark is used to apply a function to each row of a DataFrame or Dataset. It is an action that triggers the execution of the function on each element of the distributed dataset. Unlike transformations (e.g., map(), filter()), foreach() does not return a new DataFrame or Dataset but is used for side effects, such as writing data to an external system or printing rows.

​
1. Syntax

PySpark:

df.foreach(func)

Spark SQL:

  • There is no direct equivalent in Spark SQL, but you can use foreach() in DataFrame/Dataset APIs.

​
2. Parameters

  • func: A function that takes a row (or element) as input and performs an operation on it. The function can be a lambda or a user-defined function.

​
3. Key Features

  • Action: foreach() is an action, meaning it triggers the execution of the Spark job.
  • Side Effects: It is typically used for side effects, such as writing data to an external system or printing rows.
  • Distributed Execution: The function is applied to each row in a distributed manner across the cluster.

​
4. Examples

​
Example 1: Printing Each Row

PySpark:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ForeachExample").getOrCreate()

# Create DataFrame
data = [("Anand", 25), ("Bala", 30), ("Kavitha", 28), ("Raj", 35)]
columns = ["Name", "Age"]

df = spark.createDataFrame(data, columns)

# Print each row
df.foreach(lambda row: print(row))

Output:

Row(Name='Anand', Age=25)
Row(Name='Bala', Age=30)
Row(Name='Kavitha', Age=28)
Row(Name='Raj', Age=35)

​
Example 2: Writing Rows to an External System

PySpark:

# Simulate writing rows to an external system (e.g., a database)
def write_to_db(row):
    # Replace this with actual logic to write to a database
    print(f"Writing to DB: {row}")

df.foreach(write_to_db)

Output:

Writing to DB: Row(Name='Anand', Age=25)
Writing to DB: Row(Name='Bala', Age=30)
Writing to DB: Row(Name='Kavitha', Age=28)
Writing to DB: Row(Name='Raj', Age=35)

​
Example 3: Using foreach() with a User-Defined Function

PySpark:

# Define a function to process each row
def process_row(row):
    name = row["Name"]
    age = row["Age"]
    print(f"Name: {name}, Age: {age}")

# Apply the function to each row
df.foreach(process_row)

Output:

Name: Anand, Age: 25
Name: Bala, Age: 30
Name: Kavitha, Age: 28
Name: Raj, Age: 35

​
Example 4: Writing Rows to a File

PySpark:

# Simulate writing rows to a file
def write_to_file(row):
    with open("output.txt", "a") as f:
        f.write(f"{row}\n")

df.foreach(write_to_file)

Output:

  • Each row is appended to output.txt.

​
Example 5: Sending Rows to a Message Queue

PySpark:

# Simulate sending rows to a message queue
def send_to_queue(row):
    # Replace this with actual logic to send to a message queue
    print(f"Sending to queue: {row}")

df.foreach(send_to_queue)

Output:

Sending to queue: Row(Name='Anand', Age=25)
Sending to queue: Row(Name='Bala', Age=30)
Sending to queue: Row(Name='Kavitha', Age=28)
Sending to queue: Row(Name='Raj', Age=35)

​
Example 6: Using foreach() with Accumulators

PySpark:

from pyspark import AccumulatorParam

# Define a custom accumulator for strings
class StringAccumulatorParam(AccumulatorParam):
    def zero(self, initial_value):
        return initial_value
    def addInPlace(self, v1, v2):
        return v1 + v2

# Initialize an accumulator
accumulator = spark.sparkContext.accumulator("", StringAccumulatorParam())

# Use the accumulator in foreach
def accumulate_names(row):
    global accumulator
    accumulator += row["Name"] + " "

df.foreach(accumulate_names)

# Print the accumulated result
print(accumulator.value)

Output:

Anand Bala Kavitha Raj

​
5. Common Use Cases

  • Writing data to external systems (e.g., databases, message queues).
  • Logging or printing rows for debugging purposes.
  • Performing custom operations on each row (e.g., sending notifications).

​
6. Performance Considerations

  • Execution Overhead: foreach() triggers the execution of the entire DataFrame lineage, so use it carefully for large datasets.
  • Distributed Execution: The function is applied to each row in a distributed manner, so ensure the function is efficient and thread-safe.
  • Side Effects: Since foreach() is used for side effects, ensure the function does not introduce unintended behavior (e.g., modifying shared state).

​
7. Key Takeaways

  1. The foreach() function is used to apply a function to each row of a DataFrame or Dataset for side effects.
  2. It is an action that triggers the execution of the Spark job.
  3. Since foreach() is an action, it triggers the execution of the entire DataFrame lineage. Use it judiciously for large datasets.
  4. In Spark SQL, similar functionality can be achieved using foreach() in DataFrame/Dataset APIs.