Apache Spark

Apache Spark is a multi-language engine for executing data engineering, data science, and machine learning on single-node machines or large distributed clusters with Spark Connect.

When to Use

• Petabyte-Scale Distributed Data Processing: Transforming massive datasets across cloud clusters with Apache Spark 3.5+.
• Real-Time Streaming Pipelines: Low-latency stream processing with Spark Structured Streaming.
• Enterprise Lakehouse Architectures: Querying and updating Delta Lake, Apache Iceberg, and Apache Hudi tables.
• Distributed SQL & Complex Analytical Aggregations: Distributed joins, window functions, and analytics with Catalyst Optimizer.

Quick Start

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

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

df = spark.read.parquet("s3a://data-lake/sales/")
summary = df.groupBy("country").agg(avg("amount").alias("avg_spent"))
summary.write.mode("overwrite").parquet("s3a://data-lake/reports/summary/")

Core Concepts

Structured DataFrame Transformations & Window Functions

Distributed analytical querying with PySpark:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

spark = SparkSession.builder \
    .appName("CustomerAnalytics2026") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()

# Read partitioned Parquet from cloud storage
df = spark.read.parquet("s3a://data-lake/transactions/")

# Define window partitioned by customer ordered by timestamp
customer_window = Window.partitionBy("customer_id").orderBy("transaction_time")

# Compute cumulative spend and previous transaction date
enhanced_df = df.filter(F.col("status") == "COMPLETED") \
    .withColumn("running_total", F.sum("amount").over(customer_window)) \
    .withColumn("prev_transaction_time", F.lag("transaction_time", 1).over(customer_window)) \
    .withColumn("days_since_last", F.datediff(F.col("transaction_time"), F.col("prev_transaction_time")))

enhanced_df.write \
    .mode("overwrite") \
    .partitionBy("region") \
    .parquet("s3a://data-lake/marts/customer_summary/")

Broadcast Joins for Skewed Dimensions

Optimizing distributed joins between large fact tables and small dimension lookup tables:

# Broadcast small dimension table to all executors, eliminating expensive shuffles
dimension_df = spark.read.parquet("s3a://data-lake/dimensions/merchant_tiers/")

joined_df = df.join(
    F.broadcast(dimension_df),
    on="merchant_id",
    how="inner"
)

Spark Structured Streaming from Kafka

Continuous stream processing with checkpointing:

stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "telemetry-events") \
    .load()

parsed_stream = stream_df.selectExpr("CAST(value AS STRING) as json_payload")

query = parsed_stream.writeStream \
    .format("parquet") \
    .option("path", "s3a://data-lake/raw_stream/") \
    .option("checkpointLocation", "s3a://data-lake/checkpoints/telemetry/") \
    .trigger(processingTime="1 minute") \
    .start()

Common Patterns

Broadcast Join for Skewed Dimensions

Problem: Joining a small lookup table with a massive distributed fact table causes slow shuffle exchanges across executors.

Solution: Broadcast the small table to all worker nodes:

from pyspark.sql.functions import broadcast

large_fact_df = spark.read.parquet("events/")
small_dim_df = spark.read.csv("country_codes.csv", header=True)

# Broadcast join eliminates shuffle network overhead
result_df = large_fact_df.join(broadcast(small_dim_df), "country_id")

Best Practices

Do:

• Enable Adaptive Query Execution (spark.sql.adaptive.enabled=true) for automatic partition coalescing and join optimization.
• Use F.broadcast() when joining a large DataFrame with a small table (< 100MB) to eliminate shuffle overhead.
• Always specify an explicit checkpointLocation on durable cloud storage for Structured Streaming jobs.
• Cache or persist (df.persist()) intermediate DataFrames only when reused multiple times in the same job.

Don't:

• Collect massive DataFrames to the driver node (df.collect()); write directly to cloud storage.
• Use Python UDFs (@udf) if native pyspark.sql.functions exist; native functions avoid JVM-Python serialization overhead.
• Leave small file problems unaddressed; use Delta Lake OPTIMIZE or coalesce partitions before writing.

Troubleshooting

References

• Apache Spark