spark

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:

```python
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

Error	Cause	Solution
java.lang.OutOfMemoryError: Java heap space on Executor	Data skew or partition sizes too large for executor memory.	Increase spark.executor.memory or salt skewed keys before grouping.
Container killed by YARN for exceeding memory limits	Off-heap overhead or Python process exceeding memory threshold.	Increase spark.yarn.executor.memoryOverhead to 1GB+.
Total size of serialized results is bigger than spark.driver.maxResultSize	Calling .collect() on massive distributed DataFrame.	Write output to disk or use .take(100) instead of collecting entire dataset.

References

• Apache Spark