RDD - Action Foreach

The foreach(func) action applies a user-defined function to each individual element of the RDD. It is unique because it returns no value back to the Driver program (None).

Instead, it is used strictly to trigger side-effects locally on the executor worker nodessuch as writing records directly to an external database or sending events to a message queue.

Internal Execution Behavior

Because foreach() is executed directly on the executor nodes:

• No Driver Data Transfer: Data never moves over the network back to the Driver, keeping execution fast and memory-safe.
• Executor Stdout: If you call print() inside your foreach() function, the print outputs will appear inside the stdout console logs of the executor machines, not inside your Driver's terminal!

graph TD
    subgraph Driver["Driver Program (Master)"]
        DP["foreach(log_element) triggered"]
    end

    subgraph Cluster["Executors (Worker Nodes)"]
        E1["Executor 1 processes row locally"] -->|print()| Log1["Executor 1 Stdout Log"]
        E2["Executor 2 processes row locally"] -->|print()| Log2["Executor 2 Stdout Log"]
    end

    DP -->|Deploy tasks| E1
    DP -->|Deploy tasks| E2

    style Driver fill:#e1f5fe,stroke:#039be5,stroke-width:2px;
    style Cluster fill:#efebe9,stroke:#8d6e63,stroke-width:2px;

PySpark Code Example

Setup Spark Session

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("RDD Action Foreach") \
    .master("local[*]") \
    .getOrCreate()

sc = spark.sparkContext

In Action: Logging on Workers

Let's see how foreach() applies a side-effect function:

# 1. Input RDD of transaction tuples: (TxnID, Amount)
transactions = sc.parallelize([
    ("Txn501", 100), ("Txn502", 20), ("Txn503", 350)
])

# 2. Define a side-effect logging function
def process_on_executor(txn_tuple):
    txn_id, amount = txn_tuple

    # In a local test environment, prints will show in your terminal
    # In a production cluster, this prints strictly to executor log files!
    print(f"[Executor Log] Processing transaction: {txn_id} | Amount: ${amount}")

    # In production, you would write database inserts here:
    # db_client.insert(txn_id, amount)

# 3. Trigger the foreach action (lazy transformations are computed and executed!)
transactions.foreach(process_on_executor)

print("Foreach action completed successfully.")