PySpark Data Skew — Type Mismatch Breaks Broadcast Join

A fintech team I worked with spent three weeks tuning a PySpark job that aggregated transaction records for daily risk reports. It ran fine on 10 million rows in staging. At 2 billion rows in production, it silently stalled for six hours, then crashed the cluster with a java.lang.OutOfMemoryError: GC overhead limit exceeded. The root cause wasn't bad code — it was a single unpartitioned join that forced every executor to shuffle 400GB of data through a single node. One line of misunderstood API destroyed a week's worth of cluster credits.

PySpark sits at the intersection of Python's ecosystem and Apache Spark's distributed execution engine. That's a powerful combination, but it's also a trap for developers who treat it like pandas with a bigger machine. Spark doesn't run your Python the way you think it does. Your DataFrame transformations are lazy. Your joins can silently cause catastrophic data skew. Your UDFs are serialized across a JVM boundary in a way that can cut throughput by 10x. Understanding the execution model isn't academic — it's the difference between a job that completes in 8 minutes and one that runs your cloud bill into the thousands.

After this article, you'll know how to configure a SparkSession for production, write transformations that respect Spark's lazy evaluation model, tune partitioning to eliminate shuffle bottlenecks, debug skewed joins using the Spark UI, and write Spark-native aggregations instead of Python UDFs that strangle your executors. Concrete patterns. Runnable code. The failure modes that textbooks skip.

Why PySpark Tutorials Miss the Real Problem: Type Mismatch in Broadcast Joins

A PySpark tutorial typically teaches the basics of DataFrame operations, transformations, and actions. But the real-world challenge isn't syntax — it's performance. One of the most insidious performance killers is data skew caused by type mismatch in broadcast joins. When you broadcast a small table to all executors, Spark expects the join key types to match exactly. If they don't — say one side is INT and the other is STRING — Spark silently casts one side, often to STRING, which can blow up the broadcast hash table size by 2-3x due to string overhead. This turns a fast broadcast join into a memory-exhausting disaster.

In practice, a broadcast join works by sending the smaller table to every executor, where it's stored in memory as a hash map. With mismatched types, Spark inserts a cast operation that can change the hash distribution. For example, casting an INT to STRING changes the hash code, potentially causing all rows with the same integer value to hash to the same bucket. This creates severe data skew: one executor handles millions of rows while others sit idle. The join time jumps from seconds to hours, and you may see OOM errors on the skewed executor.

Use this knowledge when designing ETL pipelines that join dimension tables (often small) with fact tables (large). Always verify join key types match exactly — same data type, same precision. A simple df.withColumn("key", col("key").cast("int")) before the join can prevent hours of debugging. In production, this is the difference between a 5-minute job and a job that crashes at 2 AM.

SparkSession Setup and the Execution Model You Must Understand First

Most tutorials show you spark = SparkSession.builder.getOrCreate() and move on. That's like showing someone a car key and skipping the part about combustion engines. Before you write a single transformation, you need to understand what Spark actually does with your code — because it doesn't run it.

Spark uses lazy evaluation. Every transformation you write — filter, select, join, groupBy — builds a logical query plan. Nothing executes until you call an action: show(), count(), write(), collect(). This is why you can chain twenty transformations and Spark will optimize the entire chain before touching a single byte of data. The Catalyst optimizer reorders predicates, prunes unused columns, and sometimes rewrites your join strategy entirely. It's genuinely impressive — until you start debugging and wonder why your print statements inside a map never fire.

The DAG (Directed Acyclic Graph) is Spark's execution blueprint. Each action triggers a job, which splits into stages wherever a shuffle is required, and stages split into tasks that run in parallel across executor cores. Shuffles are expensive because they require data to move across the network between executors. Every wide transformation — groupBy, join, distinct, repartition — causes a shuffle. Narrow transformations — filter, select, withColumn — don't. This distinction drives every performance decision you'll make in production.

# io.thecodeforge — Python tutorial

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType, LongType

# Production SparkSession — never use defaults for anything beyond local testing.
# spark.sql.shuffle.partitions defaults to 200, which is catastrophically wrong
# for both small datasets (200 near-empty tasks) and massive ones (OOM per task).
spark = (
    SparkSession.builder
    .appName("transaction-risk-aggregator")
    .config("spark.sql.shuffle.partitions", "400")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.autoBroadcastJoinThreshold", "50mb")
    .getOrCreate()
)

spark.sparkContext.setLogLevel("WARN")

transaction_schema = StructType([
    StructField("transaction_id", StringType(), nullable=False),
    StructField("account_id",     StringType(), nullable=False),
    StructField("merchant_id",    StringType(), nullable=False),
    StructField("amount_usd",     DoubleType(), nullable=False),
    StructField("transaction_ts", TimestampType(), nullable=False),
    StructField("risk_score",     DoubleType(), nullable=True),
])

transaction_df = (
    spark.read
    .schema(transaction_schema)
    .option("mergeSchema", "false")
    .parquet("/data/transactions/year=2024/month=01/")
)

high_risk_transactions = (
    transaction_df
    .filter(F.col("risk_score") > 0.85)
    .filter(F.col("amount_usd") > 100.0)
    .select("transaction_id", "account_id", "merchant_id", "amount_usd", "risk_score")
    .withColumn("risk_tier",
        F.when(F.col("risk_score") > 0.95, F.lit("CRITICAL"))
         .when(F.col("risk_score") > 0.85, F.lit("HIGH"))
         .otherwise(F.lit("MEDIUM"))
    )
)

high_risk_transactions.explain(mode="formatted")
record_count = high_risk_transactions.count()
print(f"High-risk transactions in January 2024: {record_count:}")

Joins, Shuffles, and the Data Skew Problem That Tanks Production Jobs

Joins are where production PySpark jobs go to die. Not because joins are bad — because developers write them without thinking about what Spark has to do physically to execute them. When you join two DataFrames, Spark needs to get matching keys onto the same executor. That means shuffling data across the network. On a well-distributed dataset, this is fine. On a skewed dataset — where one key represents 40% of your data — one executor gets buried while the other 99 sit idle. Your job appears to be 99% complete for six hours, then either finishes three days late or crashes.

The most common skew pattern I see in production is joining on customer_id or merchant_id in transactional data. Real-world data isn't uniform. Your top merchant processes ten thousand times more transactions than your median merchant. When you join your transactions table against a merchant metadata table on merchant_id, all records for that top merchant route to a single executor. I've personally watched this kill a Spark job at a payments company at 11pm on a Friday — the job had run fine for months, then the top merchant's volume doubled during a flash sale and suddenly one task ran for four hours while 199 tasks completed in two minutes.

Spark 3.x Adaptive Query Execution (AQE) helps here, but it's not magic. You still need to understand broadcast joins, salting strategies, and when to break a complex join into multiple simpler stages.

# io.thecodeforge — Python tutorial

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, IntegerType

spark = (
    SparkSession.builder
    .appName("merchant-risk-join")
    .config("spark.sql.shuffle.partitions", "400")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256mb")
    .getOrCreate()
)

merchant_schema = StructType([
    StructField("merchant_id",       StringType(), nullable=False),
    StructField("merchant_name",     StringType(), nullable=True),
    StructField("merchant_category", StringType(), nullable=True),
    StructField("country_code",      StringType(), nullable=True),
])

transaction_schema = StructType([
    StructField("transaction_id", StringType(), nullable=False),
    StructField("account_id",     StringType(), nullable=False),
    StructField("merchant_id",    StringType(), nullable=False),
    StructField("amount_usd",     DoubleType(), nullable=False),
])

merchant_df = spark.read.schema(merchant_schema).parquet("/data/merchants/")
transaction_df = spark.read.schema(transaction_schema).parquet("/data/transactions/")

# Approach 1: Broadcast join
merchant_enriched_df = transaction_df.join(
    F.broadcast(merchant_df),
    on="merchant_id",
    how="left"
)

# Approach 2: Salting for large skewed joins
SALT_BUCKETS = 20
transaction_salted_df = transaction_df.withColumn(
    "salt",
    (F.rand() * SALT_BUCKETS).cast(IntegerType())
).withColumn(
    "salted_merchant_id",
    F.concat(F.col("merchant_id"), F.lit("_"), F.col("salt").cast(StringType()))
)

merchant_exploded_df = merchant_df.withColumn(
    "salt",
    F.explode(F.array([F.lit(i) for i in range(SALT_BUCKETS)]))
).withColumn(
    "salted_merchant_id",
    F.concat(F.col("merchant_id"), F.lit("_"), F.col("salt").cast(StringType()))
)

skew_fixed_df = (
    transaction_salted_df.join(
        merchant_exploded_df,
        on="salted_merchant_id",
        how="left"
    )
    .drop("salt", "salted_merchant_id")
)

join_miss_count = skew_fixed_df.filter(F.col("merchant_name").isNull()).count()
total_count = skew_fixed_df.count()
print(f"Total transactions enriched: {total_count:,}")
print(f"Transactions with missing merchant data: {join_miss_count:,}")
print(f"Enrichment rate: {((total_count - join_miss_count) / total_count) * 100:.2f}%")
skew_fixed_df.groupBy(F.spark_partition_id()).count().describe().show()

Aggregations Without UDFs: Why Native Spark Functions Outperform Python 10x

Here's what I see constantly from developers coming from pandas: they hit a transformation that's slightly complex, they can't immediately find the built-in Spark function, and they write a Python UDF. It feels natural. It works in testing. Then it hits production at scale and your job takes four times longer than it should.

The reason is the JVM boundary. Spark's execution engine runs on the JVM. Your Python UDFs run in a separate Python process on each executor. For every batch of rows, Spark has to serialize data from JVM memory, ship it across a local socket to the Python process, execute your Python code, serialize the results back, and deserialize them into JVM memory. This round-trip happens millions of times. I've measured a trivially simple string transformation running 8x slower as a Python UDF than as a native Spark SQL function call.

The fix is to learn pyspark.sql.functions deeply. It covers 95% of what you'd ever want to do with a UDF. Window functions handle running totals, rankings, and lag/lead calculations. Higher-order functions (transform, filter, aggregate) handle array and map columns. When you genuinely need custom logic that has no Spark equivalent, use Pandas UDFs (also called vectorized UDFs) — they batch rows into pandas DataFrames using Apache Arrow, which eliminates the per-row serialization cost and typically runs within 2x of native Spark performance.

# io.thecodeforge — Python tutorial

from pyspark.sql import SparkSession, functions as F, Window
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType, IntegerType
import pandas as pd

spark = (
    SparkSession.builder
    .appName("account-behaviour-features")
    .config("spark.sql.shuffle.partitions", "200")
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)

transaction_schema = StructType([
    StructField("transaction_id", StringType(),   nullable=False),
    StructField("account_id",     StringType(),   nullable=False),
    StructField("merchant_id",    StringType(),   nullable=False),
    StructField("amount_usd",     DoubleType(),   nullable=False),
    StructField("transaction_ts", TimestampType(), nullable=False),
])

transaction_df = spark.read.schema(transaction_schema).parquet("/data/transactions/")

account_time_window = (
    Window
    .partitionBy("account_id")
    .orderBy("transaction_ts")
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)

account_30day_window = (
    Window
    .partitionBy("account_id")
    .orderBy(F.col("transaction_ts").cast("long"))
    .rangeBetween(-30 * 24 * 3600, 0)
)

feature_df = (
    transaction_df
    .withColumn("account_lifetime_spend", F.sum("amount_usd").over(account_time_window))
    .withColumn("account_txn_rank", F.rank().over(account_time_window))
    .withColumn("rolling_30d_spend", F.sum("amount_usd").over(account_30day_window))
    .withColumn("rolling_30d_txn_count", F.count("transaction_id").over(account_30day_window))
    .withColumn("prev_transaction_amount", F.lag("amount_usd", 1).over(Window.partitionBy("account_id").orderBy("transaction_ts")))
    .withColumn("amount_delta_vs_prev", F.col("amount_usd") - F.coalesce(F.col("prev_transaction_amount"), F.lit(0.0)))
    .withColumn("is_large_transaction", F.when(F.col("amount_usd") > F.col("rolling_30d_spend") * 0.5, F.lit(True)).otherwise(F.lit(False)))
)

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType

@pandas_udf(DoubleType())
def compute_velocity_score(rolling_count: pd.Series, rolling_spend: pd.Series) -> pd.Series:
    count_score = (rolling_count / rolling_count.max()).fillna(0)
    spend_score = (rolling_spend / rolling_spend.max()).fillna(0)
    return (count_score * 0.4 + spend_score * 0.6).round(4)

scored_df = feature_df.withColumn(
    "velocity_risk_score",
    compute_velocity_score(F.col("rolling_30d_txn_count").cast(DoubleType()), F.col("rolling_30d_spend"))
)

scored_df.repartition(200, "account_id").write.mode("overwrite").partitionBy("account_id").parquet("/data/features/account_behaviour/")

output_df = spark.read.parquet("/data/features/account_behaviour/")
output_df.select("transaction_id", "account_id", "amount_usd", "rolling_30d_spend", "rolling_30d_txn_count", "is_large_transaction", "velocity_risk_score").orderBy("account_id", "transaction_ts").show(8, truncate=False)

Writing to Production Storage: Partition Strategy, Output Modes, and Avoiding the Small Files Problem

Writing Spark output correctly is just as important as reading and processing correctly, and it's where I see the most rookie mistakes land in production. The most insidious one: after all your careful processing, you write out with the default partition count — or worse, you repartition(1) because you want a single output file — and you've just created either thousands of tiny 1KB files or one massive unparallelizable blob.

The small files problem is real and painful. HDFS and object stores like S3 weren't designed for millions of tiny files. Each file carries metadata overhead. AWS S3 LIST operations are expensive and slow. Downstream Spark jobs reading your output have to open one file handle per partition file — if you wrote 10,000 partitions with 3 tasks each, your downstream reader opens 30,000 files just to start processing. I've seen a single poorly-partitioned write turn a downstream job's startup time from 8 seconds to 12 minutes.

The right approach is intentional: partition your output by the dimensions your downstream queries actually filter on, target 100-500MB per output file, and use coalesce (not repartition) when you need to reduce partition count without a shuffle. For streaming or incremental pipelines, understand the difference between overwrite, append, and the Delta Lake / Iceberg merge patterns — because overwrite on partitioned data can silently delete partitions you didn't intend to touch.

# io.thecodeforge — Python tutorial

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = (
    SparkSession.builder
    .appName("daily-risk-report-writer")
    .config("spark.sql.shuffle.partitions", "400")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.minPartitionSize", "128mb")
    .getOrCreate()
)

transaction_schema = StructType([
    StructField("transaction_id", StringType(),    nullable=False),
    StructField("account_id",     StringType(),    nullable=False),
    StructField("merchant_id",    StringType(),    nullable=False),
    StructField("amount_usd",     DoubleType(),    nullable=False),
    StructField("transaction_ts", TimestampType(), nullable=False),
    StructField("risk_tier",      StringType(),    nullable=True),
    StructField("country_code",   StringType(),    nullable=True),
])

processed_df = spark.read.schema(transaction_schema).parquet("/data/enriched_transactions/")

dated_df = processed_df.withColumn("report_date", F.to_date(F.col("transaction_ts")))
                       .withColumn("report_hour", F.hour(F.col("transaction_ts")))

current_partitions = dated_df.rdd.getNumPartitions()
print(f"Current partition count before write: {current_partitions}")

dated_df_repartitioned = dated_df.repartition(200, "report_date", "risk_tier")

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

dated_df_repartitioned.write.mode("overwrite").partitionBy("report_date", "risk_tier").option("compression", "snappy").parquet("/data/risk_reports/daily/")

# Alternative: coalesce for small summary
summary_df = processed_df.groupBy("country_code", "risk_tier", F.to_date("transaction_ts").alias("report_date"))\
    .agg(
        F.count("transaction_id").alias("transaction_count"),
        F.sum("amount_usd").alias("total_amount_usd"),
        F.avg("amount_usd").alias("avg_amount_usd"),
        F.countDistinct("account_id").alias("unique_accounts")
    ).coalesce(10)

summary_df.write.mode("overwrite").option("compression", "snappy").parquet("/data/risk_reports/daily_summary/")

output_df = spark.read.parquet("/data/risk_reports/daily_summary/")
output_df.orderBy("report_date", "country_code", "risk_tier").show(10)
print(f"Total summary records: {output_df.count():,}")

Debugging Production PySpark Jobs Using the Spark UI and Metrics

When your job fails — not if, when — you need to know exactly where to look. The Spark UI is your primary diagnostic tool. It exposes everything: DAG visualization, stage details, task metrics, executor memory, shuffle read/write, and GC time. Most engineers never open it until something breaks. You should be reading it during development.

A critical metric often missed: Shuffle Read Size / Records per task. If one task reads 10GB while others read <100MB, you have data skew. AQE can help split skewed partitions at runtime, but you must enable spark.sql.adaptive.skewJoin.enabled=true and set a reasonable threshold (e.g., 256mb).

Another essential setting for debugging: spark.sql.adaptive.logLevel=TRACE — this logs every AQE decision like partition coalescing or skew splitting. In a recent incident, that log revealed a partition was not being split because the skew threshold was set to 256MB but the actual skewed partition was 249MB. Lesson: round down your thresholds.

Finally, shuffle spill metrics tell you when data exceeds executor memory. If you see 'Shuffle Spill (Memory)' and 'Shuffle Spill (Disk)' in task metrics, your partitions are too large. Increase partition count or executor memory. Spilling to disk is a last resort — it slows jobs by orders of magnitude.

# io.thecodeforge — Python tutorial

from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder
    .appName("debugging-example")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.logLevel", "TRACE")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "128mb")
    .config("spark.sql.adaptive.coalescePartitions.initialPartitionNum", "400")
    .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "256mb")
    .config("spark.ui.port", "4040")
    .config("spark.eventLog.enabled", "true")
    .config("spark.eventLog.dir", "hdfs:///user/spark/eventLog")
    .config("spark.sql.debug", "true")
    .getOrCreate()
)

df_large = spark.range(0, 1000000, 1).withColumnRenamed("id", "key").repartition(100)
df_small = spark.range(0, 10, 1).withColumnRenamed("id", "key").repartition(10)

df_skewed = df_large.withColumn("key", F.when(F.rand() < 0.4, F.lit(9)).otherwise(F.col("key")))

result = df_skewed.join(df_small, "key", "left")
result.explain(mode="formatted")
result.count()
print("Job completed. Check Spark UI for task metrics.")

Data Cleaning and Null Handling: Why Spark's Defaults Will Corrupt Your Models

Your ML pipeline is a house of cards if you don't control how nulls propagate. Spark's default behavior? NaNs in floats, nulls everywhere else. No crash. No warning. Your aggregation silently drops rows, your feature columns become half-empty, and your model learns from ghosts.

Stop relying on na.drop() as a crutch. It's a blunt instrument that kills legitimate data. The production approach is column-level strategy: median imputation for numeric columns with <5% nulls, mode for categoricals, indicator flags so you track why data went missing.

Here's the kicker: Spark's fillna() is lazy until an action triggers evaluation. A single null in a key column used in a join will expand your record count or, worse, silently match everything. Always run a null audit before any transform. df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns]).show() — run that, fix the gaps, then proceed.

Do not drop() before you understand. You'll ship a broken dataset and debug it at 2 AM.

// io.thecodeforge — python tutorial

from pyspark.sql import SparkSession, functions as F

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

df = spark.read.parquet("s3://production-orders/raw")

// Step 1: Audit null percentages per column
null_counts = df.select([
    (F.count(F.when(F.col(c).isNull(), True)) / F.count("*") * 100).alias(f"{c}_null_pct")
    for c in df.columns
])
null_counts.show(truncate=False)
// Output:
// +--------------+--------------+
// |order_id_null |revenue_null  |
// +--------------+--------------+
// |0.0           |4.3           |
// +--------------+--------------+

// Step 2: Impute revenue with median - only 4.3% missing, safe to fill
median_revenue = df.approxQuantile("revenue", [0.5], 0.0)[0]
df_fixed = df.fillna({"revenue": median_revenue})

// Step 3: Add indicator column for rows we modified
df_final = df_fixed.withColumn("revenue_was_null", F.when(F.col("revenue").isNull(), True).otherwise(False))

Partitioning and Performance Optimization: Your Cluster Is Begging You To Stop Shuffling

Every shuffle is a cluster-wide data transfer. It's the most expensive operation in PySpark — network I/O, disk spill, serialization overhead. Your job runs 40 minutes when it should run 8. The fix? Control partitioning before Spark does it for you.

Shuffles happen on join, groupBy, distinct, orderBy. If you see a Exchange stage in the Spark UI, you're shuffling. The naive fix is to tune spark.sql.shuffle.partitions (default 200). But that's a band-aid. The real optimization is reducing the need to shuffle at all.

repartition() by your join key before the join. If you're joining two DataFrames on user_id, both should be partitioned into the same number of partitions on user_id. This becomes a map-side operation — no shuffle.

Another trick: bucketing. Use bucketBy when writing intermediate tables. Spark metadata tracks the bucket structure, so future joins on that key skip shuffles entirely. This single change cut one of my team's ETLs from 22 minutes to 3.

Use df.explain() on every join. If you see Exchange in the physical plan, you're paying a tax you didn't need to.

// io.thecodeforge — python tutorial

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("shuffle_killer") \
    .config("spark.sql.shuffle.partitions", "48") \
    .getOrCreate()

// Two DataFrames we'll join on user_id
orders = spark.read.parquet("s3://orders/2024-10-01")
users = spark.read.parquet("s3://users/active")

// Without pre-partitioning (triggers shuffle)
bad_join = orders.join(users, "user_id", "inner")
bad_join.explain()
// Physical plan shows: Exchange hashpartitioning(user_id, 200)

// With pre-partitioning: same partition count, same key
repart_orders = orders.repartition(48, "user_id")
repart_users = users.repartition(48, "user_id")
good_join = repart_orders.join(repart_users, "user_id", "inner")
good_join.explain()
// Physical plan: no Exchange in join; it's a map-only step

Filtering and Selection: Stop Dragging Dead Data Across the Cluster

Most engineers think filtering is just a WHERE clause. In PySpark, a bad filter before a join costs you thousands of dollars in shuffle time. You filter late. You filter after reading everything. You're burning cluster cycles on data you never needed.

The rule: filter as early as possible. Push predicates down to the source. If you're reading Parquet, partition pruning happens automatically when you filter on partitioned columns. If you're reading from a JDBC source, Spark can push the filter into the query — but only if you use the right predicates.

Selection is worse. Selecting all columns is a laziness tax. Every unnecessary column multiplies memory pressure during shuffles. Use select() or drop() explicitly. If you're joining two DataFrames, project down to only the columns you need before the join. Your executors will thank you.

// io.thecodeforge — python tutorial

from pyspark.sql import SparkSession

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

df = spark.read.parquet("s3://events/2024/")

# WRONG: filter after reading everything
df_bad = df.filter(df.event_type == "purchase")

# RIGHT: predicate pushdown to Parquet reader
df_good = spark.read.parquet("s3://events/2024/", date="2024-11-01")

# Push even harder with explicit filter
from pyspark.sql.functions import col
df_filtered = spark.read.parquet("s3://events/2024/") \
    .filter(col("date") == "2024-11-01") \
    .select("user_id", "revenue")

df_filtered.show(2)

Advanced PySpark Techniques: When Window Functions Beat GroupBy Hands Down

You've been taught GROUP BY for aggregations. It works. But when you need running totals, rank by partition, or lag/lead — GROUP BY forces you to lose row-level detail. That's where window functions obliterate the competition.

Window functions let you aggregate over partitioned rows while keeping every original row intact. No explode, no collect_list hack, no Python UDF. This is the difference between a 5-minute job and a 45-minute one.

With window functions, you define a partition spec and an order. Then you call rank(), row_number(), sum() over that window. Spark optimizes the shuffle automatically — one pass, no self-joins. When you add a Python UDF to a window, you break Tungsten optimization. Use native functions. Every time. The performance gap isn't small — it's 10x.

// io.thecodeforge — python tutorial

from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import rank, sum as ssum, col

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

df = spark.createDataFrame([
    ("A", "Q1", 100), ("A", "Q2", 200),
    ("B", "Q1", 150), ("B", "Q2", 250)
], ["dept", "quarter", "revenue"])

w = Window.partitionBy("dept").orderBy("quarter")

df_result = df.select(
    "*",
    rank().over(w).alias("rank"),
    ssum("revenue").over(w).alias("running_total")
)

df_result.show()

PySpark Installation Guide: Why Your Local Setup Won't Match Production

Most PySpark failures start before a single line of code runs — wrong Java version, mismatched Scala jars, or a Spark binary that refuses to talk to your Python interpreter. Production clusters ship a curated Spark distribution; your laptop likely doesn't. Install via pip install pyspark only if you accept that maven dependencies and Hadoop client binaries must align. For real work, use conda with a pinned Spark version and explicit JAVA_HOME. The root cause of "SparkSession not available" is almost always a classpath collision from multiple Spark installations. Test with a minimal broadcast join — if it silently falls back to shuffle, your installation skipped the essential pyspark-...-slim package. Dependencies: Java 8 or 11, Python 3.8+, and never mix Spark 3.4 with Python 3.12 without checking binary compatibility.

// io.thecodeforge — python tutorial

import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-11-openjdk-amd64"

from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .appName("InstallationCheck") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

df = spark.range(5)
df.show()

// Expected output: 5 rows, 0..4
// Failure means JVM or classpath mismatch

Learning PySpark From Scratch: The 3-Scan Method That Builds Real Intuition

Reading documentation top-to-bottom wastes weeks. Start with three targeted scans. First scan: run a word count on a single CSV — this exposes the lazy evaluation trap where no data moves until you call .show() or .write(). Second scan: force a broadcast join with a 10-row lookup table against 10 million rows. Watch the Spark UI see if it actually broadcasted or fell back to sort-merge (the default for small tables under 50MB). Third scan: debug a null count. Default Spark drops nulls in aggregations silently — your model just lost 2% of records. After these three exercises, you understand partitioning, shuffle, and null semantics better than someone who read three books. Next concrete step: take your own production query, run EXPLAIN on it, and trace every Exchange node in the physical plan. That single skill separates engineers who tune jobs from those who guess.

// io.thecodeforge — python tutorial

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

# Scan 1: Lazy execution test
df = spark.read.csv("sales.csv", header=True)
print("No action yet")

# Scan 2: Broadcast join
dim = spark.range(100).toDF("id")
facts = spark.range(1_000_000).toDF("id")
joined = facts.join(dim.hint("broadcast"), "id")

// Scan 3: Null trap
data = spark.createDataFrame([(1,), (None,)], ["val"])
print(data.agg({"val": "count"}).collect())
# Shows 1 — null was dropped
// Expected: [Row(count(val)=1)]