df = df.withColumn("a", expr("..."))
df = df.withColumn("b", expr("..."))
df = df.withColumn("c", expr("..."))
# ... 50 more times

df = df.select(
    "*",
    expr("...").alias("a"),
    expr("...").alias("b"),
    expr("...").alias("c"),
    # all at once
)

from functools import reduce
transforms = [("a", expr("...")), ("b", expr("...")), ("c", expr("..."))]
df = reduce(lambda d, t: d.withColumn(t[0], t[1]), transforms, df)
# Still creates nested nodes but cleaner code. For extreme cases, use select().

from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_df), "key")

from pyspark.sql.functions import lit, rand, floor, explode, array, col

salt_buckets = 10

# Salt the large (skewed) side
large_df = large_df.withColumn("salt", floor(rand() * salt_buckets).cast("int"))

# Replicate the small side for each salt value
small_df = small_df.withColumn(
    "salt", explode(array([lit(i) for i in range(salt_buckets)]))
)

# Join on key + salt
result = large_df.join(small_df, ["key", "salt"]).drop("salt")

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")

# Identify skewed keys
skewed_keys = ["key1", "key2"]  # from analysis

# Process skewed keys with broadcast join
skewed_result = large_df.filter(col("key").isin(skewed_keys)) \
    .join(broadcast(small_df), "key")

# Process non-skewed keys normally
normal_result = large_df.filter(~col("key").isin(skewed_keys)) \
    .join(small_df, "key")

# Combine
result = skewed_result.union(normal_result)

from pyspark.sql.functions import concat, lit, floor, rand, split, sum as _sum

# Phase 1: Partial aggregation with salt
salted = df.withColumn("salted_key", concat(col("key"), lit("_"), floor(rand() * 100).cast("string")))
partial = salted.groupBy("salted_key").agg(_sum("value").alias("partial_sum"))

# Phase 2: Remove salt, final aggregation
result = partial.withColumn("key", split(col("salted_key"), "_")[0]) \
    .groupBy("key").agg(_sum("partial_sum").alias("total_sum"))

from pyspark.sql import Window
from pyspark.sql.functions import row_number, rank, dense_rank

w = Window.partitionBy("department").orderBy(col("salary").desc())

from pyspark.sql import Window
from pyspark.sql.functions import sum as _sum, avg, col

# Running total (all rows up to current)
cumulative_w = Window.partitionBy("category").orderBy("date") \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)

# 7-day moving average (current + 6 prior rows)
moving_w = Window.partitionBy("category").orderBy("date") \
    .rowsBetween(-6, Window.currentRow)

# Total for percentage (all rows in partition)
total_w = Window.partitionBy("category")

result = df.select(
    "*",
    _sum("revenue").over(cumulative_w).alias("running_total"),
    avg("revenue").over(moving_w).alias("moving_avg_7d"),
    (col("revenue") / _sum("revenue").over(total_w) * 100).alias("pct_of_total")
)

from pyspark.sql.functions import unix_timestamp

# Convert date to seconds for rangeBetween
days_7 = 7 * 86400  # 7 days in seconds
w = Window.partitionBy("category") \
    .orderBy(unix_timestamp("date")) \
    .rangeBetween(-days_7, 0)

from pyspark.sql import Window
from pyspark.sql.functions import first, last, lead, datediff, col, row_number

w = Window.partitionBy("customer_id").orderBy("purchase_date")

result = df.withColumn("purchase_rank", row_number().over(w)) \
    .withColumn("first_purchase", first("purchase_date").over(
        w.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
    )) \
    .withColumn("last_purchase", last("purchase_date").over(
        w.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
    )) \
    .withColumn("next_purchase", lead("purchase_date", 1).over(w)) \
    .filter(col("purchase_rank") == 1) \
    .withColumn("days_to_second_purchase",
        datediff(col("next_purchase"), col("purchase_date"))
    )

# Co-locate all rows with same user_id on same partition
df = df.repartition(100, "user_id")
# Now a groupBy("user_id") or join on "user_id" won't need shuffle

df.write.bucketBy(256, "user_id").sortBy("user_id").saveAsTable("bucketed_users")

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType
import pandas as pd

@pandas_udf(DoubleType())
def normalize(s: pd.Series) -> pd.Series:
    return (s - s.mean()) / s.std()

df = df.withColumn("normalized_salary", normalize(col("salary")))

from pyspark.sql.functions import pandas_udf, PandasUDFType

@pandas_udf(output_schema, PandasUDFType.GROUPED_MAP)
def train_model(pdf: pd.DataFrame) -> pd.DataFrame:
    # Train a model per group
    model = LinearRegression().fit(pdf[features], pdf[target])
    pdf["prediction"] = model.predict(pdf[features])
    return pdf

result = df.groupBy("category").apply(train_model)

@pandas_udf(DoubleType(), PandasUDFType.GROUPED_AGG)
def weighted_mean(values: pd.Series, weights: pd.Series) -> float:
    return (values * weights).sum() / weights.sum()

result = df.groupBy("category").agg(weighted_mean(col("value"), col("weight")))

def predict_batch(iterator):
    import pickle
    model = pickle.load(open("/dbfs/models/model.pkl", "rb"))
    for batch_df in iterator:
        batch_df["prediction"] = model.predict(batch_df[feature_cols])
        yield batch_df

result = spark_df.mapInPandas(predict_batch, schema=output_schema)

30 minutes") \
    .groupBy(
        col("user_id"),
        window("event_time", "1 hour")  # 1-hour tumbling window
    ).count()">from pyspark.sql.functions import window

clicks = spark.readStream.format("kafka").load().select(
    col("user_id"),
    col("event_time").cast("timestamp"),
    col("page_url")
)

# Define watermark: accept data up to 30 min late
sessionized = clicks \
    .withWatermark("event_time", "30 minutes") \
    .groupBy(
        col("user_id"),
        window("event_time", "1 hour")  # 1-hour tumbling window
    ).count()

2 hours")
payments = payments_stream.withWatermark("payment_time", "3 hours")

# Time-range condition limits state
joined = orders.join(
    payments,
    expr("""
        orders.order_id = payments.order_id AND
        payments.payment_time BETWEEN orders.order_time AND orders.order_time + interval 1 hour
    """),
    "left_outer"
)"># Both streams must have watermarks
orders = orders_stream.withWatermark("order_time", "2 hours")
payments = payments_stream.withWatermark("payment_time", "3 hours")

# Time-range condition limits state
joined = orders.join(
    payments,
    expr("""
        orders.order_id = payments.order_id AND
        payments.payment_time BETWEEN orders.order_time AND orders.order_time + interval 1 hour
    """),
    "left_outer"
)

1 minute") \
    .start()">def upsert_to_delta(batch_df, batch_id):
    target = DeltaTable.forName(spark, "silver_orders")

    target.alias("t").merge(
        batch_df.alias("s"),
        "t.order_id = s.order_id"
    ).whenMatchedUpdateAll() \
     .whenNotMatchedInsertAll() \
     .execute()

spark.readStream.table("bronze_orders") \
    .writeStream \
    .foreachBatch(upsert_to_delta) \
    .option("checkpointLocation", "/checkpoints/silver_orders") \
    .trigger(processingTime="1 minute") \
    .start()

def idempotent_write(batch_df, batch_id):
    # Use batch_id to ensure idempotency
    batch_df.write.format("delta") \
        .mode("overwrite") \
        .option("replaceWhere", f"batch_id = {batch_id}") \
        .save("/path/to/output")

spark.conf.set(
    "spark.sql.streaming.stateStore.providerClass",
    "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider"
)

class StateMonitor(StreamingQueryListener):
    def onQueryProgress(self, event):
        state_info = event.progress.stateOperators
        for op in state_info:
            print(f"State rows: {op.numRowsTotal}, Memory: {op.memoryUsedBytes}")

from pyspark.sql.functions import row_number, col
from pyspark.sql import Window

window = Window.partitionBy("id").orderBy(col("updated_at").desc())
deduped = df.withColumn("rn", row_number().over(window)) \
    .filter(col("rn") == 1) \
    .drop("rn")

# Always pass distinct values list to avoid an extra job
pivoted = df.groupBy("date") \
    .pivot("category", ["Electronics", "Clothing", "Food"]) \
    .agg(sum("amount"))

from pyspark.sql.functions import expr

# Spark 3.4+ native unpivot:
unpivoted = df.unpivot("date", ["Electronics", "Clothing", "Food"], "category", "amount")

# Before 3.4 — use stack:
unpivoted = df.selectExpr(
    "date",
    "stack(3, 'Electronics', Electronics, 'Clothing', Clothing, 'Food', Food) as (category, amount)"
)

from pyspark.sql.functions import lead
from pyspark.sql import Window

w = Window.orderBy("sequence_id")
gaps = df.withColumn("next_id", lead("sequence_id").over(w)) \
    .filter(col("next_id") - col("sequence_id") > 1) \
    .select(
        col("sequence_id").alias("gap_start"),
        col("next_id").alias("gap_end"),
        (col("next_id") - col("sequence_id") - 1).alias("gap_size")
    )

30 minutes in seconds

df = df.withColumn("prev_time", lag("event_time").over(w)) \
    .withColumn("new_session",
        when(
            (unix_timestamp("event_time") - unix_timestamp("prev_time")) > session_timeout, 1
        ).when(col("prev_time").isNull(), 1)  # first event is always new session
        .otherwise(0)
    ) \
    .withColumn("session_id",
        _sum("new_session").over(w.rowsBetween(Window.unboundedPreceding, Window.currentRow))
    )">from pyspark.sql.functions import lag, when, unix_timestamp, sum as _sum, monotonically_increasing_id
from pyspark.sql import Window

w = Window.partitionBy("user_id").orderBy("event_time")
session_timeout = 30 * 60  # 30 minutes in seconds

df = df.withColumn("prev_time", lag("event_time").over(w)) \
    .withColumn("new_session",
        when(
            (unix_timestamp("event_time") - unix_timestamp("prev_time")) > session_timeout, 1
        ).when(col("prev_time").isNull(), 1)  # first event is always new session
        .otherwise(0)
    ) \
    .withColumn("session_id",
        _sum("new_session").over(w.rowsBetween(Window.unboundedPreceding, Window.currentRow))
    )

from pyspark.sql.functions import explode, col

raw = spark.read.json("/path/to/nested.json")

# Assume structure: {id, orders: [{order_id, amount, items: [{product, qty}]}]}
flat = raw.select("id", explode("orders").alias("order")) \
    .select(
        "id",
        col("order.order_id"),
        col("order.amount"),
        explode("order.items").alias("item")
    ).select(
        "id",
        "order_id",
        "amount",
        col("item.product").alias("product_name"),
        col("item.qty").alias("quantity")
    )

from pyspark.sql.types import StructType, ArrayType

def flatten_df(df):
    """Recursively flatten all nested structs and arrays."""
    flat_cols = []
    for field in df.schema.fields:
        if isinstance(field.dataType, StructType):
            for subfield in field.dataType.fields:
                flat_cols.append(col(f"{field.name}.{subfield.name}").alias(f"{field.name}_{subfield.name}"))
        elif isinstance(field.dataType, ArrayType):
            df = df.withColumn(field.name, explode(col(field.name)))
            return flatten_df(df)  # Recurse after explode
        else:
            flat_cols.append(col(field.name))
    return df.select(flat_cols)

from pyspark.sql.functions import dense_rank, col, sum as _sum
from pyspark.sql import Window

# First aggregate revenue per product per category
product_revenue = df.groupBy("category", "product_id") \
    .agg(_sum("revenue").alias("total_revenue"))

# Rank within each category
w = Window.partitionBy("category").orderBy(col("total_revenue").desc())
top3 = product_revenue.withColumn("rnk", dense_rank().over(w)) \
    .filter(col("rnk") <= 3) \
    .drop("rnk")

from pyspark.sql.functions import lag, col
from pyspark.sql import Window

w = Window.partitionBy("sensor_id").orderBy("timestamp")
result = df.withColumn("prev_value", lag("value", 1).over(w)) \
    .withColumn("delta", col("value") - col("prev_value"))

from pyspark.sql.functions import avg, col
from pyspark.sql import Window

w = Window.partitionBy("department")
result = df.withColumn("dept_avg", avg("salary").over(w)) \
    .filter(col("salary") > col("dept_avg")) \
    .drop("dept_avg")

# Method 1: left_anti join (most efficient)
loyal_customers = orders_df.join(returns_df, "customer_id", "left_anti") \
    .select("customer_id").distinct()

# Method 2: left_outer + filter
loyal_customers = orders_df.join(returns_df, "customer_id", "left_outer") \
    .filter(returns_df["return_id"].isNull()) \
    .select(orders_df["customer_id"]).distinct()

from pyspark.sql.functions import lag, col, round as _round
from pyspark.sql import Window

w = Window.partitionBy("product_id").orderBy("month")

growth = monthly_revenue.withColumn("prev_revenue", lag("revenue", 1).over(w)) \
    .withColumn("mom_growth_pct",
        _round(
            (col("revenue") - col("prev_revenue")) / col("prev_revenue") * 100, 2
        )
    )

from pyspark.sql.functions import collect_set, explode, col, array_sort
from itertools import combinations

# Get all products per order
order_products = df.groupBy("order_id") \
    .agg(collect_set("product_id").alias("products"))

# Explode into pairs (using UDF since built-in is limited)
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import ArrayType, StructType, StructField, StringType
import pandas as pd

@pandas_udf(ArrayType(StringType()))
def get_pairs(products: pd.Series) -> pd.Series:
    return products.apply(lambda x: [f"{a}|{b}" for a, b in combinations(sorted(x), 2)])

pairs = order_products.withColumn("pair", explode(get_pairs(col("products")))) \
    .groupBy("pair").count() \
    .orderBy(col("count").desc())

# Use PERMISSIVE mode with corrupt record column
df = spark.read.option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .schema(expected_schema) \
    .csv("/path/to/data.csv")

# Separate good and bad
good_records = df.filter(col("_corrupt_record").isNull()).drop("_corrupt_record")
bad_records = df.filter(col("_corrupt_record").isNotNull()) \
    .select("_corrupt_record")

# Write both
good_records.write.format("delta").mode("append").saveAsTable("silver_data")
bad_records.write.format("delta").mode("append").saveAsTable("quarantine_data")

from pyspark.sql.functions import percentile_approx, expr

# Approximate median (fast, good enough for most cases)
result = df.groupBy("department") \
    .agg(percentile_approx("salary", 0.5).alias("median_salary"))

# Exact median using window function
from pyspark.sql.functions import count, row_number, col, avg
from pyspark.sql import Window

w = Window.partitionBy("department").orderBy("salary")
total_w = Window.partitionBy("department")

result = df.withColumn("rn", row_number().over(w)) \
    .withColumn("cnt", count("*").over(total_w)) \
    .filter(
        (col("rn") == (col("cnt") / 2).cast("int") + 1) |
        ((col("cnt") % 2 == 0) & (col("rn") == (col("cnt") / 2).cast("int")))
    ) \
    .groupBy("department") \
    .agg(avg("salary").alias("median_salary"))

10 minutes") \
    .dropDuplicates(["event_id"])

# Write to Delta with foreachBatch for MERGE-based dedup
def upsert_events(batch_df, batch_id):
    from delta.tables import DeltaTable
    if DeltaTable.isDeltaTable(spark, "/delta/events"):
        target = DeltaTable.forPath(spark, "/delta/events")
        target.alias("t").merge(
            batch_df.alias("s"), "t.event_id = s.event_id"
        ).whenNotMatchedInsertAll().execute()
    else:
        batch_df.write.format("delta").save("/delta/events")

deduped.writeStream \
    .foreachBatch(upsert_events) \
    .option("checkpointLocation", "/checkpoints/events") \
    .trigger(processingTime="30 seconds") \
    .start()">from pyspark.sql.functions import from_json, col, expr

# Define schema
schema = "event_id STRING, user_id STRING, event_type STRING, event_time TIMESTAMP, payload STRING"

# Read from Kafka
raw = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker1:9092") \
    .option("subscribe", "events") \
    .option("startingOffsets", "latest") \
    .load()

# Parse
parsed = raw.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

# Deduplicate using watermark
deduped = parsed \
    .withWatermark("event_time", "10 minutes") \
    .dropDuplicates(["event_id"])

# Write to Delta with foreachBatch for MERGE-based dedup
def upsert_events(batch_df, batch_id):
    from delta.tables import DeltaTable
    if DeltaTable.isDeltaTable(spark, "/delta/events"):
        target = DeltaTable.forPath(spark, "/delta/events")
        target.alias("t").merge(
            batch_df.alias("s"), "t.event_id = s.event_id"
        ).whenNotMatchedInsertAll().execute()
    else:
        batch_df.write.format("delta").save("/delta/events")

deduped.writeStream \
    .foreachBatch(upsert_events) \
    .option("checkpointLocation", "/checkpoints/events") \
    .trigger(processingTime="30 seconds") \
    .start()