> ## Documentation Index
> Fetch the complete documentation index at: https://rajanand.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Spark: foreach function

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:**

```python theme={"system"}
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:**

```python theme={"system"}
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:**

```python theme={"system"}
# 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:**

```python theme={"system"}
# 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:**

```python theme={"system"}
# 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:**

```python theme={"system"}
# 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:**

```python theme={"system"}
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.