Spark & Big Data

RDDs (Resilient Distributed Datasets) are the foundational abstraction: type-safe, unstructured, JVM-based collections with full control over physical execution but no optimizer. DataFrames are distributed tables with named columns and schema, processed through the Catalyst query optimizer for automatic query planning and optimization; available in Python, R, Scala, and Java. Datasets combine DataFrame optimizations with compile-time type safety (Scala/Java only); Python has no Dataset API. For Python ML workloads, DataFrames (via PySpark) are the standard. DataFrames and Datasets execute identically on Spark's optimized engine; RDDs bypass Catalyst and are slower for structured data.

# DataFrame API (preferred)
df = spark.read.parquet('s3://bucket/data')
result = df.groupBy('category').agg(F.sum('revenue'))

Lazy evaluation means Spark builds a logical query plan from transformation calls (map, filter, groupBy, join) without executing them immediately. Execution only happens when an action is triggered (collect, count, write, show). This deferred execution allows Catalyst to analyze the full query plan holistically, apply logical optimizations (predicate pushdown, projection pruning, join reordering), generate multiple physical plan candidates, and select the cheapest using a cost model. Lazy evaluation also prevents unnecessary computation — if a filter eliminates 90% of data, that elimination is applied before expensive joins. This makes Spark's query planner fundamentally more powerful than eager row-by-row execution frameworks.

# Transformations are lazy — no execution yet
df = spark.read.parquet('s3://data')
filtered = df.filter(df.amount > 1000)  # plan built
joined = filtered.join(users, 'user_id')  # plan extended
joined.write.parquet('s3://output')  # execution triggers here

Narrow transformations compute each output partition using data from exactly one input partition, requiring no data movement across the network. Examples: map, filter, flatMap, union, sample. Wide transformations (also called shuffle transformations) require data from multiple input partitions to compute each output partition, necessitating a shuffle — network data exchange between executors. Examples: groupByKey, reduceByKey, join (without broadcast), distinct, repartition, sortBy. Shuffles are expensive: they write data to disk, serialize/deserialize, and transfer over the network. Minimizing shuffles is the primary Spark performance optimization: prefer reduceByKey over groupByKey, use broadcast joins for small tables, and filter data early.

Catalyst is Spark SQL's extensible query optimizer built on Scala's functional pattern matching. It transforms a logical query plan through four phases: Analysis (resolving column names against catalog), Logical Optimization (applying rule-based rewrites: predicate pushdown, constant folding, projection pruning, null propagation), Physical Planning (generating multiple physical plans and estimating costs via the CBO if statistics are available), and Code Generation (Tungsten whole-stage code generation that compiles query stages to JVM bytecode). Key rules include predicate pushdown (filter before join), partition pruning (skip irrelevant Parquet files), join reordering (CBO), and converting broadcast-eligible joins automatically when the smaller table fits in the broadcast threshold.

spark.conf.set('spark.sql.cbo.enabled', 'true')
# Show execution plan
df.filter(df.amount > 100).join(users, 'user_id').explain(True)

AQE (Spark 3.0+, enabled by default in 3.2+) re-optimizes query plans at runtime using statistics collected during execution, addressing cases where static pre-execution optimization is insufficient. Three key capabilities: (1) dynamically coalescing shuffle partitions to eliminate tiny post-shuffle partitions, reducing task overhead; (2) dynamically switching join strategies — converting sort-merge joins to broadcast joins if runtime data reveals one side is smaller than the broadcast threshold; (3) dynamically optimizing skew joins by splitting skewed partitions into smaller sub-partitions. AQE reduces the need for manual spark.sql.shuffle.partitions tuning, which is a major source of performance issues in production Spark jobs.

spark.conf.set('spark.sql.adaptive.enabled', 'true')
spark.conf.set('spark.sql.adaptive.skewJoin.enabled', 'true')
spark.conf.set('spark.sql.adaptive.coalescePartitions.enabled', 'true')

A broadcast join sends the entire smaller table (the broadcast side) to every executor, converting a sort-merge join (which requires shuffling both sides) into a map-side join with zero shuffle. This eliminates the expensive shuffle of the large table. Spark automatically broadcasts tables smaller than spark.sql.autoBroadcastJoinThreshold (default 10 MB). Force a broadcast with broadcast() hint or F.broadcast(). Use broadcast joins when one side is clearly small (dimension tables, lookup tables, filter lists). Do not broadcast tables approaching executor memory limits — the table is materialized on every executor, multiplying memory usage by the number of executor cores. Disable with spark.sql.autoBroadcastJoinThreshold=-1 if causing OOM.

from pyspark.sql import functions as F
result = large_df.join(
    F.broadcast(small_lookup_df),
    on='product_id'
)

Spark's persist() supports multiple StorageLevels: MEMORY_ONLY stores deserialized JVM objects in memory (fastest access, highest memory use). MEMORY_AND_DISK spills to disk when memory is full (safe default). MEMORY_ONLY_SER stores serialized (compact) objects in memory (slower access, less memory). MEMORY_AND_DISK_SER combines serialization with disk spillover. DISK_ONLY persists to disk only (slowest). OFF_HEAP stores in off-heap memory (avoids GC pressure). The _2 suffix replicates partitions on two nodes for fault tolerance. Use cache() (equivalent to MEMORY_AND_DISK in PySpark) for DataFrames reused multiple times in a query or across actions. Unpersist cached data with df.unpersist() after it is no longer needed.

from pyspark import StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK)
df.count()  # triggers caching
# Later:
df.unpersist()

repartition(n) performs a full shuffle to produce exactly n evenly distributed partitions. It is expensive but produces balanced partitions and can increase or decrease partition count. Use before heavy computations requiring balanced parallelism, or after filtering that left few records in many partitions. coalesce(n) merges partitions without a full shuffle by combining adjacent partitions on the same executor. It can only reduce partition count; attempting to increase partitions with coalesce raises an error or falls back to repartition. Use coalesce before writing output to reduce the number of output files. A common pattern: coalesce(1) to write a single CSV — be aware this collects all data to one executor and risks OOM on large datasets.

# Increase or rebalance
df = df.repartition(200, 'date_col')  # partition by date
# Reduce before write (no shuffle)
df.coalesce(10).write.parquet('s3://output')

Data skew occurs when partitions have vastly unequal sizes, causing some tasks to take orders of magnitude longer than others (the straggler problem). It commonly appears with NULL keys, hot user IDs, or uneven category distributions in groupBy and join operations. Solutions: (1) AQE's dynamic skew join optimization (Spark 3.0+) automatically splits skewed partitions; (2) Salting — append a random integer to the skewed key before joining, replicating the smaller side; (3) Broadcast join if the non-skewed side is small; (4) Pre-aggregate before join to reduce data volume; (5) Increase spark.sql.adaptive.skewJoin.skewedPartitionFactor to tune detection sensitivity. Identify skew via Spark UI's Stage detail showing max vs. median task duration gap.

import pyspark.sql.functions as F
# Salting technique for skewed joins
salt = 10
large_df = large_df.withColumn('salt', (F.rand()*salt).cast('int'))
small_df = small_df.withColumn('salt', F.explode(F.array([F.lit(i) for i in range(salt)])))
result = large_df.join(small_df, on=['key', 'salt'])

Structured Streaming treats a live data stream as an unbounded table that continuously grows. Queries on this table produce a result table updated incrementally. The engine runs micro-batches (or continuous processing mode for sub-millisecond latency) against the incoming data. Key abstractions: input source (Kafka, file system, socket), output sink (Kafka, Parquet, console, Delta Lake), output mode (append adds new rows; complete replaces entire result; update modifies changed rows), trigger (interval, once, availableNow), and watermark for handling late-arriving events. State management enables stateful operations like windowed aggregations and stream-stream joins with watermarking. Checkpointing to reliable storage enables fault tolerance and exactly-once semantics.

stream = spark.readStream.format('kafka') \
    .option('kafka.bootstrap.servers', 'broker:9092') \
    .option('subscribe', 'events').load()
stream.writeStream.format('delta') \
    .option('checkpointLocation', '/ckpt') \
    .trigger(processingTime='30 seconds') \
    .start()

Watermarking allows the engine to bound the state maintained for windowed aggregations and stream-stream joins by defining a maximum allowed lateness for events. Set with df.withWatermark('event_time', '10 minutes'): this tells Spark to wait up to 10 minutes past the maximum observed event_time before finalizing window results. Events arriving more than 10 minutes late are dropped. The watermark advances as the maximum event_time in incoming data increases. Without watermarking, windowed aggregations keep state indefinitely, causing unbounded memory growth. Watermarking is required to use append output mode with aggregations (otherwise only update or complete modes are allowed). Tune the threshold based on source latency characteristics.

windowed = stream \
    .withWatermark('event_time', '10 minutes') \
    .groupBy(
        F.window('event_time', '5 minutes'),
        'user_id'
    ).count()

Window functions compute values over a sliding window of rows defined by PARTITION BY (the grouping), ORDER BY (ordering within the partition), and a frame specification (ROWS BETWEEN or RANGE BETWEEN). Common functions: rank(), dense_rank(), row_number() for ranking; lag(), lead() for accessing adjacent rows; sum(), avg() over windows for running aggregations. In PySpark, use Window.partitionBy().orderBy() to define the window spec. Performance tips: add partitionBy columns to the query's partitioning to reduce shuffle; avoid unbounded frames (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) on large partitions; cache the base DataFrame if multiple window functions are applied. Unbounded RANGE frames with ORDER BY on float columns can be unpredictable.

from pyspark.sql import Window
w = Window.partitionBy('user_id').orderBy('event_time')
df = df.withColumn('rank', F.rank().over(w)) \
       .withColumn('prev_event', F.lag('event_type', 1).over(w))

A UDF (User-Defined Function) allows applying custom Python logic to DataFrame columns. However, Python UDFs have a serious performance penalty: Spark must serialize each row from JVM to Python process via the Py4J bridge, execute Python, and deserialize results back. This row-by-row overhead can be 10-100x slower than native Spark SQL expressions. Alternatives: (1) Pandas UDFs (vectorized UDFs) process batches of rows as Pandas Series/DataFrames via Apache Arrow, reducing serialization overhead dramatically; (2) Spark SQL built-in functions (F.regexp_extract, F.to_timestamp, etc.) always preferred when available; (3) Spark SQL expressions registered with spark.udf.register for use in SQL strings. Always exhaust built-in functions before resorting to Python UDFs.

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

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

df = df.withColumn('norm_score', normalize(df.score))

Delta Lake is an open-source storage layer that adds ACID transactions, schema enforcement, schema evolution, time travel (versioned snapshots), and DML operations (UPDATE, DELETE, MERGE/upsert) to Parquet-based data lakes. It maintains a transaction log (_delta_log directory) recording every change as a JSON commit entry, enabling point-in-time reads (TIMESTAMP AS OF or VERSION AS OF) and rollback. Without Delta, Parquet lakes lack atomicity — partial writes leave corrupt data; no DELETE or UPDATE support requires full table rewrites. Delta's MERGE INTO statement implements efficient upserts (SCD Type 1) without rewriting unchanged data. Z-ordering and liquid clustering optimize data layout for predicate-selective queries.

# Upsert with Delta MERGE
delta_table.alias('t').merge(
    updates_df.alias('s'), 't.id = s.id'
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

Spark executor memory (set via spark.executor.memory) is divided into three regions. Reserved memory (300 MB by default) is for Spark internal objects and is not configurable. Usable memory (executor.memory - 300 MB) is divided by spark.memory.fraction (default 0.6) into execution + storage memory (Unified Memory) versus user memory (0.4 fraction). Within Unified Memory, spark.memory.storageFraction (default 0.5) grants storage (cache) a soft guaranteed share, though execution memory can evict cached data under pressure. User memory (40% of usable) holds user data structures, UDF objects, and RDD metadata. spark.executor.memoryOverhead (default 10% or 384 MB) provides off-heap overhead for JVM overhead and native memory used by Python processes.

spark.conf.set('spark.executor.memory', '8g')
spark.conf.set('spark.executor.memoryOverhead', '2g')
spark.conf.set('spark.memory.fraction', '0.75')

A Spark Job is triggered by an action (collect, write, count). Each job is divided into Stages separated by shuffle boundaries — wide transformations create new stage boundaries. Within each stage, tasks are the atomic unit of parallel execution: one task per input partition. All tasks within a stage can run in parallel across executor cores. After each shuffle, the next stage's tasks read the shuffled data. The number of tasks in a stage equals the number of partitions in that stage's data. Understanding this hierarchy is essential for debugging: the Spark UI shows jobs, stages, and per-task metrics. Long stages with few tasks indicate insufficient partitioning; stages with many tiny tasks indicate over-partitioning.

Parquet optimization in Spark involves multiple techniques. Partition pruning: partition files on disk by date or category; filters on partition columns skip entire directories without reading data. Predicate pushdown: Catalyst pushes row-level filters into the Parquet reader, skipping row groups that fail min/max statistics. Column pruning: select only needed columns; Parquet's columnar format reads only requested columns from disk. File size: aim for 128-512 MB Parquet files to match HDFS/S3 block sizes; many small files cause excessive task overhead. Schema evolution: use mergeSchema=True carefully as it increases parse time. Enable Parquet statistics collection (spark.sql.parquet.filterPushdown=true, enabled by default). Use vectorized Parquet reader (spark.sql.parquet.enableVectorizedReader=true).

# Efficient Parquet read with pruning
df = spark.read.parquet('s3://lake/events') \
    .filter(F.col('dt') == '2025-01-15') \
    .select('user_id', 'event_type', 'amount')

MLlib is Spark's distributed machine learning library built on DataFrames. The Pipeline API uses two abstractions: Transformers (stateless) apply a transform() method to add or modify columns (StringIndexer output, VectorAssembler, Normalizer). Estimators (stateful) implement fit() on training data, producing a fitted Transformer model (LogisticRegressionModel, StandardScalerModel). A Pipeline chains Estimators and Transformers sequentially; calling pipeline.fit(train_df) fits each stage in order. The result is a PipelineModel that applies all stages in transform(). CrossValidator wraps a Pipeline for distributed hyperparameter tuning across a parameter grid. MLlib supports classification, regression, clustering, collaborative filtering, and feature engineering at scale.

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler, StandardScaler
pipeline = Pipeline(stages=[
    VectorAssembler(inputCols=['f1','f2'], outputCol='features'),
    StandardScaler(inputCol='features', outputCol='scaled'),
    LogisticRegression(featuresCol='scaled', labelCol='label')
])
model = pipeline.fit(train_df)

Spark on Kubernetes runs the driver as a Kubernetes pod and dynamically provisions executor pods via the Kubernetes API, then terminates them when done. There is no persistent resource manager — Kubernetes itself handles scheduling. Benefits: native cloud integration, pod-level resource limits, container image-based dependency management, and integration with K8s secrets and ConfigMaps. Spark on YARN requires a persistent Hadoop cluster with the YARN ResourceManager managing long-lived NodeManager daemons. YARN provides better data locality with HDFS. Kubernetes suits cloud-native ML platforms (Kubeflow, Spark Operator) with ephemeral compute. YARN suits on-premise Hadoop deployments with HDFS data locality optimization. Dynamic allocation works differently: Kubernetes terminates executor pods while YARN releases slots.

spark-submit \
  --master k8s://https://k8s-api:6443 \
  --deploy-mode cluster \
  --conf spark.kubernetes.container.image=spark:3.5.0 \
  local:///app/job.py

Dynamic partition overwrite (spark.sql.sources.partitionOverwriteMode=dynamic, or mode('overwrite') on partitioned DataFrames) overwrites only the partitions present in the write DataFrame, leaving unaffected partitions intact. Without this, mode('overwrite') on a partitioned table deletes the entire table before writing — a destructive and dangerous behavior in incremental load pipelines. Enable dynamic overwrite to safely update a specific date partition without touching historical data. This is essential for idempotent daily loads to partitioned Hive/Delta tables: the pipeline can be re-run for a specific date without side effects. Delta Lake's MERGE INTO is a more sophisticated alternative that handles row-level upserts rather than full partition replacement.

spark.conf.set('spark.sql.sources.partitionOverwriteMode', 'dynamic')
df.write.mode('overwrite').partitionBy('dt').parquet('s3://lake/events')

spark.sql.shuffle.partitions controls the number of partitions after a shuffle operation (groupBy, join, etc.), defaulting to 200. This single default is rarely optimal: for small datasets, 200 partitions causes massive overhead with tiny tasks; for terabyte-scale joins, 200 partitions produces huge individual partitions that spill to disk. General guideline: target 100-200 MB per post-shuffle partition. Compute as: total_data_size_GB * 1024 / target_partition_MB. With AQE enabled, Spark dynamically coalesces post-shuffle partitions at runtime, making the initial value less critical. Disable AQE coalescing (spark.sql.adaptive.coalescePartitions.enabled=false) only when manual tuning is preferred for reproducibility in benchmarks.

# Without AQE: manual tuning
spark.conf.set('spark.sql.shuffle.partitions', '800')
# With AQE: let Spark auto-coalesce
spark.conf.set('spark.sql.adaptive.enabled', 'true')

Broadcast variables allow efficiently distributing large read-only data (a lookup dictionary, a model, a list) to all executors once, avoiding repeated serialization with each task. Create with sc.broadcast(value); access with bcast.value. Without broadcast, Spark serializes the variable into every task closure — costly for large objects. Accumulators are write-only variables for aggregating values across tasks back to the driver (e.g., counting errors, summing statistics). Workers can only add to accumulators (+=); only the driver reads the final value. Custom accumulators subclass AccumulatorParam. Note: accumulators in transformations (lazy code) may be updated multiple times during re-computation; use them in actions for reliable counting.

# Broadcast
lookup = sc.broadcast({'cat_A': 1, 'cat_B': 2})
df.rdd.map(lambda r: lookup.value.get(r.category, -1))

# Accumulator
errors = sc.accumulator(0)
df.foreach(lambda r: errors.add(1) if r.val is None else None)

Speculative execution (spark.speculation=true) detects straggler tasks that run significantly slower than the median task in the same stage and launches speculative copies on other executors. The first copy to finish wins; the slower copy is killed. This reduces stage completion time when stragglers are caused by hardware anomalies (slow disk, noisy neighbor) rather than data skew. Enable on cloud environments where instance degradation is common. Disable for non-idempotent tasks that cause side effects (writing to a database, calling external APIs) since speculative execution may run the task twice. Tune spark.speculation.multiplier (default 1.5) and spark.speculation.quantile (default 0.75) to control straggler detection sensitivity.

spark.conf.set('spark.speculation', 'true')
spark.conf.set('spark.speculation.multiplier', '2.0')
spark.conf.set('spark.speculation.quantile', '0.9')

JVM garbage collection pressure is a major cause of Spark task pauses and OOM errors, particularly in ML workloads that cache large DataFrames or use UDFs. Key settings: use G1GC (-XX:+UseG1GC) instead of the default Parallel GC for better pause time control on large heaps. Set G1 region size (-XX:G1HeapRegionSize=16m) for multi-gigabyte executor heaps. Set -XX:InitiatingHeapOccupancyPercent=35 to trigger GC earlier and prevent long full-GC pauses. Use off-heap memory (spark.memory.offHeap.enabled=true) to move storage cache off the JVM heap entirely, dramatically reducing GC pressure. Monitor GC time in the Spark UI Executor tab — if GC time exceeds 10% of task time, tuning is needed.

spark.conf.set('spark.executor.extraJavaOptions',
    '-XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:InitiatingHeapOccupancyPercent=35')
spark.conf.set('spark.memory.offHeap.enabled', 'true')
spark.conf.set('spark.memory.offHeap.size', '4g')

When reading JSON or CSV with inferSchema=True, Spark samples a fraction of the data (samplingRatio, default 1.0 for JSON) to guess column types. For large files this is expensive: it requires an extra full scan pass. Risks include incorrect type inference from sample anomalies (all-integer samples inferred as LongType when DoubleType is correct), mismatch if the sample doesn't represent all data distributions, and schema instability across runs if data changes. Best practice: define an explicit schema using StructType for production pipelines. For CSV, provide schema=explicit_schema in read options. Explicit schemas also eliminate the extra sampling pass, improving read startup time. Use spark-avro or Parquet formats that embed schema for schema-on-write guarantees.

from pyspark.sql.types import *
schema = StructType([
    StructField('user_id', LongType()),
    StructField('amount', DoubleType()),
    StructField('ts', TimestampType())
])
df = spark.read.schema(schema).json('s3://input/')

Tungsten is Spark's physical execution engine that bypasses JVM overhead for critical computations. Its three main components: (1) off-heap memory management with a custom memory allocator that stores binary data without JVM object headers, reducing GC pressure and enabling cache-efficient processing; (2) cache-aware algorithms that arrange data to maximize CPU cache hits during sort and hash operations; (3) whole-stage code generation (WSCG) that compiles entire query stages into a single JVM function using the Janino library, eliminating virtual function dispatch overhead from the Volcano iterator model. WSCG is the most impactful: it converts interpreted expression evaluation (calling .evaluate() per row per expression) into compiled tight loops, achieving throughput close to hand-written Java.

# Verify WSCG in explain
df.groupBy('col').count().explain()
# Look for: WholeStageCodegen in plan output

Use the kafka format in readStream and writeStream. For reading, specify kafka.bootstrap.servers, subscribe (topic name or comma-separated list), startingOffsets ('latest' or 'earliest' or JSON offset spec), and failOnDataLoss. The resulting DataFrame has columns: key (binary), value (binary), topic, partition, offset, timestamp. Cast value to StringType and parse JSON with from_json. For writing, produce to Kafka by selecting key (optional binary) and value (binary) columns and writing with writeStream.format('kafka'). Exactly-once semantics require idempotent producer config and transactional writes. Offsets are tracked in the checkpoint location, enabling resume after failure without reprocessing or data loss.

df = spark.readStream.format('kafka') \
    .option('kafka.bootstrap.servers', 'broker:9092') \
    .option('subscribe', 'clicks') \
    .option('startingOffsets', 'latest').load()
df = df.withColumn('data', F.from_json(F.col('value').cast('string'), schema))

Spark SQL allows executing SQL queries against DataFrames by registering them as temporary views or using the Hive metastore. createOrReplaceTempView(name) registers a DataFrame as a session-scoped table accessible via spark.sql(). createOrReplaceGlobalTempView(name) creates a cluster-scoped view under the global_temp database accessible across sessions. Temporary views exist only for the duration of the SparkSession. Execute SQL with spark.sql('SELECT ...'). The result is a DataFrame on which further transformations can be chained. Spark SQL and DataFrame API are fully interchangeable — they compile to the same physical plan. Use SQL for complex analytics queries where SQL readability is preferable; use DataFrame API for programmatic transformations in ML pipelines.

df.createOrReplaceTempView('events')
result = spark.sql("""
    SELECT user_id, COUNT(*) as cnt, SUM(amount) as total
    FROM events
    WHERE dt >= '2025-01-01'
    GROUP BY user_id
    ORDER BY total DESC
""")

Z-ordering (also called Z-order clustering) reorders data within Delta Lake files so that related values (for filtered columns) are co-located in the same files, making file skipping more effective. Use OPTIMIZE table ZORDER BY (col1, col2). Spark rewrites data files to maximize the correlation between Z-order key values and file boundaries. Subsequent queries filtering on Z-order columns skip more files because Parquet min/max statistics eliminate irrelevant files. Z-ordering is effective for high-cardinality filter columns like user_id or product_id that cannot be used as partition keys (too many unique values would create too many directories). Unlike partitioning, Z-order handles multi-dimensional filtering efficiently. Databricks introduced Liquid Clustering as a more adaptive successor to Z-ordering.

# Delta SQL
OPTIMIZE events_table ZORDER BY (user_id, event_type)

Spark provides df.na methods for null handling. df.na.drop(how='any', subset=['col1','col2']) removes rows with nulls in specified columns. df.na.fill({'col1': 0, 'col2': 'unknown'}) fills nulls with specified values per column. df.na.replace(0, None) replaces specific values. More sophisticated: use Imputer from pyspark.ml.feature for mean/median/mode imputation (stateful transformer fitted on training data). For feature engineering, use withColumn with when/otherwise to create indicator columns (null_flag) before imputing, preserving the missingness signal. Always fit imputers on training data only and apply the fitted model to validation/test sets to prevent data leakage. Check null counts with df.select([F.count(F.when(F.col(c).isNull(), c)).alias(c) for c in df.columns]).

from pyspark.ml.feature import Imputer
imputer = Imputer(inputCols=['age','salary'], outputCols=['age_imp','salary_imp'], strategy='median')
model = imputer.fit(train_df)
train_df = model.transform(train_df)
test_df = model.transform(test_df)

The Spark UI (default port 4040, or History Server for completed jobs) provides detailed execution telemetry. Key tabs: Jobs shows all jobs with duration and task counts. Stages lists all stages with aggregate task metrics — sort by median/max task duration to find stragglers. Tasks within a stage shows per-task breakdown: input bytes, shuffle read/write, GC time, duration. Storage shows cached DataFrames and their memory usage. Executors shows per-executor memory, GC time, and active tasks — identify unbalanced load. SQL shows the executed physical plans with operator-level timing. Event Timeline shows scheduling delays. Common debugging: high shuffle spill → increase executor memory or reduce shuffle partitions; high GC time → tune JVM GC settings; straggler tasks → investigate data skew.

Dynamic allocation (spark.dynamicAllocation.enabled=true) allows Spark to add executors when tasks are queued and remove idle executors to free cluster resources. The scheduler adds executors exponentially (1, 2, 4, 8...) when tasks wait longer than spark.dynamicAllocation.schedulerBacklogTimeout (default 1s). Executors idle longer than spark.dynamicAllocation.executorIdleTimeout (60s default) are removed. Benefits: cost efficiency in shared clusters and cloud environments where idle executors waste money. Trade-offs: executor acquisition latency adds overhead for bursty workloads; evicting executors loses cached data (mitigated with external shuffle service). External shuffle service (spark.shuffle.service.enabled=true) must be enabled on YARN or Kubernetes for safe executor removal while preserving shuffle data.

spark.conf.set('spark.dynamicAllocation.enabled', 'true')
spark.conf.set('spark.dynamicAllocation.minExecutors', '2')
spark.conf.set('spark.dynamicAllocation.maxExecutors', '50')
spark.conf.set('spark.shuffle.service.enabled', 'true')

Checkpointing in Structured Streaming writes the query's state (streaming offset progress, stateful aggregation state) to reliable storage (HDFS, S3, GCS) at regular intervals. It enables two guarantees: (1) fault recovery — if the query fails, restart from the last checkpoint without reprocessing data; (2) exactly-once semantics when combined with an idempotent sink. Set via .option('checkpointLocation', '/reliable/path/') on the writeStream. Checkpoint data includes the source offsets consumed, the state store for aggregations, and the metadata about completed micro-batches. Without checkpointing, a restart from 'latest' loses processed offsets and may produce duplicate or missing records depending on source retention. Never share checkpoint directories between different streaming queries.

query = stream.writeStream \
    .format('delta') \
    .outputMode('append') \
    .option('checkpointLocation', 's3://bucket/checkpoints/query1') \
    .start('/delta/output')

Stream-stream joins combine two streaming sources. Both sides must have watermarks defined to bound state size — without watermarks, Spark buffers all historical records indefinitely, causing unbounded memory growth. Define watermarks on both streams (stream_a.withWatermark('ts', '30 minutes')), then join using inner, left_outer, or right_outer join type on the key and an optional event-time range condition (event times within a window of each other). The engine maintains state for each stream within the watermark bounds and produces results when matching records arrive within the allowed time range. Stream-stream joins are stateful: the state stores are checkpointed for fault tolerance. Left semi and full outer joins have limitations in streaming mode.

a = stream_a.withWatermark('ts_a', '30 minutes')
b = stream_b.withWatermark('ts_b', '30 minutes')
joined = a.join(b,
    F.expr("a_id = b_id AND ts_b >= ts_a - interval 30 minutes AND ts_b <= ts_a + interval 30 minutes")
)

Spark OOM errors occur in different memory regions: Executor OOM (Java heap space) typically from caching too much data, large shuffles materializing huge partitions, or UDF returning very large objects. Executor OOM (GC overhead limit exceeded) indicates repeated full GCs with insufficient heap. Driver OOM from collect() or toPandas() pulling too much data to the driver — always use limit() before collecting large DataFrames. Container killed by YARN/Kubernetes (OOM Killer) indicates off-heap memory (overhead) is exceeded — increase spark.executor.memoryOverhead. Diagnosis steps: check Spark UI Executors tab for GC time and memory usage, check stderr logs for java.lang.OutOfMemoryError stack traces, review Stage details for shuffle spill metrics (spill to disk indicates memory pressure).

# Common fixes
# 1. Reduce data with filter/select early
# 2. Increase executor memory
spark.conf.set('spark.executor.memory', '16g')
spark.conf.set('spark.executor.memoryOverhead', '4g')
# 3. Enable off-heap cache
spark.conf.set('spark.memory.offHeap.enabled', 'true')

Apache Arrow is an in-memory columnar data format that dramatically accelerates data exchange between Spark's JVM and Python processes. Enable with spark.conf.set('spark.sql.execution.arrow.pyspark.enabled', 'true'). This affects: toPandas() and createDataFrame(pandas_df) — conversion is 10-100x faster using Arrow's zero-copy columnar batches instead of row-by-row Py4J serialization. Pandas UDFs (vectorized UDFs) require Arrow and automatically batch rows into pandas Series/DataFrames. Arrow is also used by Spark Connect (Spark 3.4+) for client-server communication. Prerequisite: pip install pyarrow. Set spark.sql.execution.arrow.maxRecordsPerBatch to tune batch size (default 10000); increase for throughput, decrease if individual batches OOM.

spark.conf.set('spark.sql.execution.arrow.pyspark.enabled', 'true')
# Fast conversion
pandas_df = spark_df.toPandas()  # uses Arrow batches
spark_df = spark.createDataFrame(pandas_df)  # fast reverse

Spark supports several physical join algorithms selected by the Catalyst optimizer. Broadcast Hash Join: broadcasts the smaller side as a hash map to all executors; requires one side below the broadcast threshold. Sort Merge Join: the default for large-large joins; sorts both sides by join key then merges, requiring a shuffle. Hash Join (Shuffle Hash Join): builds a hash table from the smaller shuffled side; more memory-intensive than SMJ. Nested Loop Join: a last-resort O(n*m) join for non-equi joins (range conditions). Catalyst chooses based on table sizes, available statistics, and configuration. Force a join strategy with hints: BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL. AQE can switch from SMJ to broadcast join dynamically based on runtime statistics.

# Force join strategy with hints
result = large_df.hint('MERGE').join(medium_df, 'key')
# Or broadcast hint
result = large_df.join(F.broadcast(small_df), 'key')

Spark Connect decouples the Spark client from the Spark cluster using a gRPC-based client-server protocol. The client (Python, Scala, Go, or other languages) builds a logical query plan locally and sends it to the Spark Connect server over gRPC. The server resolves the plan against the actual catalog and executes it on the Spark cluster. Results are streamed back using Apache Arrow. Benefits: thin clients that don't embed the Spark runtime; upgradable Spark versions without client code changes; polyglot support (any language with gRPC); reduced client memory usage since no SparkContext runs locally. Enables IDE debugging, notebook-based development with remote clusters, and lighter containerized clients in ML serving scenarios.

# Spark Connect client (remote)
from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .remote('sc://spark-cluster:15002') \
    .getOrCreate()

Spark MLlib provides a rich set of feature transformers. StringIndexer converts categorical strings to numeric indices. OneHotEncoder creates binary vectors from indices. VectorAssembler combines multiple columns into a single feature vector. StandardScaler and MinMaxScaler normalize numeric features. Binarizer thresholds numeric values. PCA performs dimensionality reduction. HashingTF and IDF for text TF-IDF features. Word2Vec for dense word embeddings. NGram for n-gram extraction. ChiSqSelector for feature selection. Tokenizer and RegexTokenizer for text splitting. Chain transformers in a Pipeline to apply all steps consistently across train/test splits. Always fit transformers only on training data to prevent leakage, then apply to test data using the fitted model's transform().

from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
stages = [
    StringIndexer(inputCol='category', outputCol='cat_idx'),
    OneHotEncoder(inputCol='cat_idx', outputCol='cat_vec'),
    VectorAssembler(inputCols=['cat_vec','amount'], outputCol='features')
]

groupByKey shuffles all values for each key across the network to one reducer before any aggregation, potentially moving massive amounts of data. reduceByKey applies a local combine step (mini-reducer) on each executor before shuffling, sending only the partially aggregated result per key-executor combination — dramatically reducing shuffle data volume. For sum, count, min, max, and other associative-commutative aggregations, always prefer reduceByKey, aggregateByKey, or combineByKey. groupByKey is necessary only when the full list of values per key must be materialized (e.g., joining all records for a user for analysis). On DataFrames, this distinction is handled internally by the optimizer — groupBy().agg() uses partial aggregation automatically.

# Avoid:
rdd.groupByKey().mapValues(sum)  # shuffles all values
# Prefer:
rdd.reduceByKey(lambda a, b: a + b)  # local combine + shuffle reduced data

Resource management in shared Spark clusters uses several mechanisms. YARN queues (fair scheduler or capacity scheduler) partition cluster resources between teams. Spark executor limits (spark.executor.instances or dynamic allocation bounds) cap per-application resource use. Application priorities and preemption policies allow high-priority jobs to reclaim resources. Ganglia, Prometheus + Grafana, or the cluster manager UI expose cluster-wide metrics. For Kubernetes, resource quotas and limit ranges at the namespace level bound Spark application resources. Per-job CPU and memory requests (spark.executor.cores, spark.executor.memory) should reflect actual workload needs, not maximums. Use Spark History Server for post-hoc analysis of completed job resource consumption patterns across teams.

spark-submit \
  --conf spark.executor.instances=20 \
  --conf spark.executor.cores=4 \
  --conf spark.executor.memory=8g \
  --conf spark.yarn.queue=ml-team \
  my_job.py

Delta Lake's transaction log records every write operation with a version number. Time travel queries read historical snapshots without data duplication. Syntax in Spark SQL: SELECT * FROM events VERSION AS OF 42 (version number) or TIMESTAMP AS OF '2025-01-15 00:00:00'. In PySpark: spark.read.format('delta').option('versionAsOf', 42).load(path). Use cases: audit and compliance queries on historical state, reproducible ML training on consistent dataset snapshots, debugging unexpected data changes, and recovering accidentally deleted records. The RESTORE command reverts the table to a previous version. Delta retains history files per DELTA.logRetentionDuration (default 30 days) and data files per deletedFileRetentionDuration. VACUUM removes old files outside the retention window.

# Time travel read
df = spark.read.format('delta') \
    .option('timestampAsOf', '2025-01-15') \
    .load('/delta/events')

-- SQL syntax
SELECT * FROM events TIMESTAMP AS OF '2025-01-15'

Small file problems (thousands of files under 10 MB) severely degrade Spark performance: each file requires separate S3/HDFS metadata calls, task overhead dominates computation time, and listing directories is slow. Solutions: (1) Compact on write — use repartition() or coalesce() before writing to control output file count, targeting 128-512 MB files; (2) Periodic compaction — run OPTIMIZE on Delta tables (merges small files into large ones without changing data); (3) Auto-optimize in Databricks — auto-compaction and optimized writes enabled automatically; (4) Merge small upstream writes with a periodic compaction Spark job; (5) Use Delta Lake's OPTIMIZE command with Z-ordering as a dual-purpose compaction; (6) Configure spark.sql.files.maxPartitionBytes (default 128 MB) to prevent over-splitting on read.

# Write with controlled file count
df.repartition(50).write.mode('append').parquet('s3://lake/events/dt=2025-01-15')
# Delta compaction
OPTIMIZE events_table WHERE dt >= '2025-01-01'

Spark join hints override the Catalyst optimizer's automatic join strategy selection. Available hints: BROADCAST (df.hint('broadcast')) forces a broadcast hash join; MERGE forces sort-merge join; SHUFFLE_HASH forces shuffle hash join (builds hash table on smaller shuffled side); SHUFFLE_REPLICATE_NL forces nested loop join for non-equi conditions. Override the optimizer when it makes incorrect decisions due to stale statistics: forcing BROADCAST when the optimizer underestimates a table's size, or forcing MERGE when SHUFFLE_HASH causes OOM on a large right side. Collect table statistics with ANALYZE TABLE for better automatic decisions. After upgrading Spark versions, re-verify that optimizer decisions remain correct — CBO improvements can change join selections.

# Force broadcast join via hint
result = large.join(medium.hint('broadcast'), on='id')

-- Collect statistics for CBO
ANALYZE TABLE events COMPUTE STATISTICS FOR ALL COLUMNS

Trigger.AvailableNow() (Spark 3.3+) runs a streaming query in micro-batches until all currently available data is processed, then stops — behaving like a batch job with streaming semantics. Unlike Trigger.Once() (deprecated), AvailableNow processes data in multiple micro-batches (respecting maxFilesPerTrigger and maxBytesPerTrigger limits) rather than one large batch, enabling better parallelism and fault tolerance. Use this pattern for Lambda architecture pipelines that need incremental Parquet/Delta processing on a schedule: run every hour via Airflow, processing only new files since the last checkpoint. The checkpoint tracks consumed offsets so reruns are idempotent. This replaces complex custom watermarking logic with streaming's built-in offset management.

query = stream.writeStream \
    .format('delta') \
    .trigger(availableNow=True) \
    .option('checkpointLocation', '/ckpt/incremental') \
    .start('/delta/output')
query.awaitTermination()

The Cost-Based Optimizer (CBO) uses table and column statistics to generate more optimal query plans. Without CBO, Catalyst uses rule-based heuristics (e.g., always broadcast the left table in a join). With CBO enabled, it estimates cardinalities and selectivities based on collected statistics to choose better join strategies and orderings. Enable with spark.conf.set('spark.sql.cbo.enabled', 'true') and collect statistics with ANALYZE TABLE tablename COMPUTE STATISTICS FOR ALL COLUMNS. CBO improves multi-table join ordering, choosing to process smaller intermediate results first. It requires up-to-date statistics — stale stats (after large data additions without re-analysis) can cause worse plans than rule-based optimization. Most impactful for complex star-schema queries with 5+ tables.

spark.conf.set('spark.sql.cbo.enabled', 'true')
spark.sql('ANALYZE TABLE sales COMPUTE STATISTICS FOR ALL COLUMNS')
spark.sql('ANALYZE TABLE customers COMPUTE STATISTICS FOR ALL COLUMNS')

Accumulators are Spark's mechanism for aggregating values from worker tasks back to the driver, supporting add-only semantics (workers write, driver reads). Built-in accumulators support Long, Double, and Collection types; custom accumulators extend AccumulatorV2. Common misuse: (1) Relying on accumulators for correctness rather than debugging — if a stage is re-executed (due to failure or speculative execution), the accumulator is incremented again, producing double-counted results; (2) Using accumulators in transformations (lazy code) rather than actions — the accumulator updates may not happen if the transformation is not materialized; (3) Reading accumulator value before the action completes. Safe pattern: use accumulators only for monitoring (counting bad records, tracking progress) where approximate counts are acceptable, not for exact business logic.

error_count = sc.accumulator(0)
def process(row):
    if row.value is None:
        error_count.add(1)
    return row
df.foreach(process)
print(f'Bad records (approx): {error_count.value}')

Data locality minimizes network data transfer by scheduling tasks on nodes that already hold the required data. Spark's locality levels from best to worst: PROCESS_LOCAL (data in the same executor JVM process — cached RDD partitions), NODE_LOCAL (data on the same physical node — HDFS local blocks or another executor on the same host), NO_PREF (data has no locality preference — e.g., external storage), RACK_LOCAL (data on a node in the same network rack), and ANY (no locality — data must be fetched over the network). Spark waits up to spark.locality.wait (default 3s) at each level before degrading to the next. With YARN on HDFS clusters, NODE_LOCAL is common; on cloud object storage (S3/GCS), data locality is meaningless (always ANY), so maximizing parallelism rather than locality matters more.

spark.conf.set('spark.locality.wait', '5s')  # wait longer for better locality
# Check locality in Spark UI: Stages -> Task Details -> Locality Level column

The external shuffle service (ESS) runs as a persistent daemon on each worker node (separate from executor processes) that stores and serves shuffle data on behalf of executors. When dynamic allocation is enabled without ESS, removing idle executors also destroys their shuffle data — subsequent stages that need that data fail. ESS keeps shuffle data alive even after the producing executor is removed, enabling safe dynamic allocation. On YARN, the ESS is the YARN shuffle service (spark.shuffle.service.enabled=true). On Kubernetes, a dedicated ESS pod per node fulfills this role (kubernetes-shuffle-service or the Remote Shuffle Service). ESS also reduces executor GC pressure by offloading shuffle data management to a separate process.

# Enable ESS on YARN
spark.conf.set('spark.shuffle.service.enabled', 'true')
spark.conf.set('spark.dynamicAllocation.enabled', 'true')

The Spark UI Event Timeline (in the Stage detail view) shows a per-executor Gantt chart of task scheduling, execution, and garbage collection. Debugging steps: (1) Identify the slowest stage via the Stages tab (sort by Duration); (2) In the stage detail, check median vs. max task duration — a large gap indicates data skew; (3) Check the Event Timeline for scheduler delay (blue bar) — large values mean tasks queue up faster than executors process them; (4) Check deserialization time — excessive deserialization suggests large closure objects being serialized; (5) Check shuffle read/write spill (to disk) — memory pressure; (6) Check GC time per task — excessive GC (>10% of task duration) indicates memory issues. Cross-reference with the Executors tab to identify executors with abnormally high GC times or shuffle spill.