PySpark supports various control statements to manage the flow of your Spark applications. PySpark supports using Python’s if-else-elif statements, but with limitations.

Supported Usage

1. Conditional statements within PySpark scripts.
2. Controlling flow of Spark operations.
3. Dynamic filtering or data manipulation.

Unsupported Usage

• Within Spark UDFs (User-Defined Functions).
• Directly within Spark SQL queries.
• As part of Spark DataFrame operations.
• Conditional Logic in PySpark: Using if, elif, else

Conditional statements in Pyspark

1.Python’s if elif else

1. if: Execute a block of code if a condition is true.
2. elif: Execute a block of code if the initial condition is false and another condition is true.
3. else: Execute a block of code if all conditions are false.

if spark.sparkContext.getConf().get("spark.executor.instances") == "4":
    print("Using 4 executor instances")
elif spark.sparkContext.getConf().get("spark.executor.instances") == "2":
    print("Using 2 executor instances")
else:
    print("Default configuration")

from pyspark.sql import SparkSession

# Create SparkSession
spark = SparkSession.builder.appName("Conditional Example").getOrCreate()

# Conditional statement
if spark.sparkContext.getConf().get("spark.executor.instances") == "4":
    print("Using 4 executor instances")
elif spark.sparkContext.getConf().get("spark.executor.instances") == "2":
    print("Using 2 executor instances")
else:
    print("Default configuration")

# Dynamic filtering
filter_column = "name"
if filter_column == "name":
    spark.sql("SELECT * FROM customers WHERE name = 'John'")
elif filter_column == "age":
    spark.sql("SELECT * FROM customers WHERE age > 30")

2. Use when and otherwise for Simple Conditions:

In PySpark, the when and otherwise functions replace traditional if-else logic. They are chainable, meaning you can achieve the effect of elif as well.

Syntax

from pyspark.sql.functions import when, col

df.withColumn("new_column", when(condition, value_if_true).otherwise(value_if_false))

Syntax
Python
from pyspark.sql.functions import when, col

df.withColumn("new_column", when(condition, value_if_true).otherwise(value_if_false))

Example:

Here’s how you would implement an if-elif-else logic using when and otherwise:

Python if-else code (NOT supported in PySpark):

if score >= 70:
    grade = 'A'
elif score >= 50:
    grade = 'B'
else:
    grade = 'C'

Equivalent PySpark code:

pythonCopy codefrom pyspark.sql import functions as F

df = df.withColumn("grade", 
                   F.when(F.col("score") >= 70, "A")
                    .when(F.col("score") >= 50, "B")
                    .otherwise("C"))
df.show()

This will create a new column grade based on the value of the score column.

you can use when and otherwise in various PySpark scenarios beyond withColumn and aggregate functions.

Scenarios

1. select statement

df.select(
    df.name,
    when(df.age > 30, "adult").otherwise("minor").alias("age_group")
)

2. where clause

df.where(when(df.country == "USA", df.sales > 1000).otherwise(df.sales > 500))

3. join conditions

df1.join(
    df2,
    when(df1.id == df2.id, df1.id).otherwise(df1.id_null).equalTo(df2.id)
)

4. case statements (Spark SQL)

spark.sql("""
    SELECT *,
    CASE
        WHEN age > 30 THEN 'adult'
        ELSE 'minor'
    END AS age_group
    FROM customers
""")

5. filter method

df.filter(when(df.country == "USA", df.sales > 1000).otherwise(df.sales > 500))

6. union method

df.union(
    df2.select(
        when(df2.id == df.id, df2.name).otherwise(df.name).alias("name")
    )
)

7. withColumnRenamed

df.withColumnRenamed(
    "age",
    when(df.age > 30, "adult_age").otherwise("minor_age")
)

Tips and Variations

1. Chain multiple when statements for complex conditions.
2. Use otherwise to specify default values.
3. Combine with other PySpark functions (e.g., and, or).
4. Leverage Spark SQL’s CASE statement for more complex logic.

3. Using UDFs (User Defined Functions) for Complex Logic:

If you need more complex logic in your conditions that cannot easily be handled by when() and otherwise(), you can use UDFs to apply Python if-else logic to the DataFrame.

Example:

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# Define the Python function
def grade(score):
    if score >= 70:
        return 'A'
    elif score >= 50:
        return 'B'
    else:
        return 'C'

# Convert the Python function to a UDF
grade_udf = udf(grade, StringType())

# Apply the UDF to the DataFrame
df = df.withColumn("grade", grade_udf(F.col("score")))
df.show()

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def age_group(age):
    if age > 30:
        return "adult"
    else:
        return "minor"

age_group_udf = udf(age_group, StringType())
customers_df.withColumn("age_group", age_group_udf("age"))

Here, the function grade behaves like a Python if-elif-else statement and is applied to the DataFrame using udf.

4. Using F.expr() for SQL-like Conditions:

Another way to handle conditional logic is by writing SQL-like expressions using F.expr().

Example:

df = df.withColumn(
    "grade", 
    F.expr("CASE WHEN score >= 70 THEN 'A' WHEN score >= 50 THEN 'B' ELSE 'C' END")
)
df.show()

This CASE WHEN syntax is equivalent to if-elif-else in SQL and is supported within PySpark’s F.expr() function.

5.Spark SQL queries: Use CASE statements or IF functions.

spark.sql("SELECT *, CASE WHEN age > 30 THEN 'adult' ELSE 'minor' END AS age_group

Summary:

• Direct Python if-else or elif statements cannot be used in PySpark DataFrame operations because PySpark processes data across a cluster in a distributed fashion.
• Use when and otherwise for conditional logic in PySpark DataFrames.
• For more complex logic, use UDFs (User Defined Functions).
• SQL-style CASE WHEN expressions are supported through F.expr() for more complex conditions.

Looping in PySpark

Looping Over Multiple Transformation Steps in PySpark

If you need to loop over multiple transformation steps and apply them, you can use a Python loop to apply the transformations step by step.

for: Iterate over a sequence (e.g., list, tuple).
while: Execute a block of code while a condition is true.
Python
# For loop
data = [1, 2, 3, 4, 5]
for num in data:
    print(num)

# While loop
i = 0
while i < 5:
    print(i)
    i += 1

Example:

# Example: Loop over a list of transformations
transformation_steps = [
    ("score", F.col("score") * 2),
    ("name", F.upper(F.col("name")))
]

for col_name, transformation in transformation_steps:
    df = df.withColumn(col_name, transformation)

df.show()

Looping in PySpark: Using foreach, collect, or map

You cannot directly use Python for loops on a PySpark DataFrame. However, you can use PySpark’s distributed operations like foreach(), collect(), or map() to process each row or iterate over columns.

Example of a loop using foreach():

You can use foreach() to apply a function to each row of a DataFrame.

# Define a function to apply to each row
def process_row(row):
    print(f"Name: {row['name']}, Score: {row['score']}")

# Apply the function using foreach
df.foreach(process_row)

This function will print each row, but it does not modify the DataFrame. foreach() is used for side effects, such as writing to external systems.

Example of looping over columns:

If you want to loop over columns and apply some operation, you can use a simple Python for loop to iterate over the column names.

# Example: trim whitespace from all string columns
from pyspark.sql.types import StringType

# Loop through columns and apply trim if the column is of StringType
for col_name, col_type in df.dtypes:
    if col_type == 'string':
        df = df.withColumn(col_name, F.trim(F.col(col_name)))

df.show()

Looping Over Data in PySpark Using collect()

You can use collect() to bring the entire DataFrame to the driver node and then iterate over it. This is not recommended for large datasets, as it can lead to memory issues.

Example:

# Collect the DataFrame to the driver as a list of rows
rows = df.collect()

# Loop through the collected data
for row in rows:
    print(f"Name: {row['name']}, Score: {row['score']}")

Python control statements like if-else can still be used in PySpark when they are applied in the context of driver-side logic, not in DataFrame operations themselves.

Here’s how the logic works in your example:

Understanding Driver-Side Logic in PySpark

• Driver-Side Logic: The driver is the main program running your PySpark code. Python’s native control structures (like if-else and loops) can be applied on the driver side, which controls what operations to trigger. This means that if-else statements can control the flow of your Spark job, but they do not operate directly on the distributed data across the cluster.
• PySpark Operations: When you’re working with distributed data (i.e., DataFrames or RDDs), you need to use PySpark’s API to apply control logic on that data. The transformations and actions in PySpark are evaluated lazily, meaning they don’t actually run until an action (like count() or show()) triggers them.

Breakdown of Your Example

Driver-Side if-else Statement: In the following part of the code:                                       if spark.sparkContext.getConf().get("spark.executor.instances") == "4": print("Using 4 executor instances") elif spark.sparkContext.getConf().get("spark.executor.instances") == "2": print("Using 2 executor instances") else: print("Default configuration")

This if-else statement works because it is evaluated on the driver (the main control point of your Spark application). It is checking the Spark configuration and printing the appropriate message based on the value of the spark.executor.instances setting.

These are decisions made at the driver level, not within the distributed computation on the worker nodes.

Dynamic Filtering with SQL: filter_column = "name" if filter_column == "name": spark.sql("SELECT * FROM customers WHERE name = 'John'") elif filter_column == "age": spark.sql("SELECT * FROM customers WHERE age > 30")

This if-else block is also evaluated on the driver. It chooses which SQL query to execute based on the value of the filter_column variable.

The actual query (spark.sql()) will be distributed across the cluster, but the decision on which query to run is controlled by the if-else logic on the driver side.

Summary

• Yes, you can use Python’s if-else statements in PySpark, but they are only applicable on the driver side (for controlling which Spark operation gets executed).
• When you’re working with transformations on DataFrames (which are distributed across the cluster), you need to use PySpark-specific functions like when, filter, select, etc.
• Driver-side logic: You are not missing anything! Driver-side logic (like checking configuration, deciding which DataFrame to create, or which SQL to run) is perfectly valid in PySpark.

The confusion often arises because PySpark DataFrames themselves operate in a distributed fashion, and thus require different control structures for operations on the data itself (like the when–otherwise functions I mentioned earlier). But outside of that, normal Python control flow still works for guiding the structure of your Spark job!