pyspark-and-apache-spark-analytics

name	pyspark-and-apache-spark-analytics
description	Master production PySpark and Apache Spark analytics, distributed data pipelines, Delta Lake/Iceberg optimization, memory management, and skew resolution.

PySpark & Apache Spark Analytics

Comprehensive engineering guide for architecting, optimizing, and operating large-scale distributed data processing pipelines using PySpark and Apache Spark 3.x/4.x.

---

1. Core Principles & Architecture

1.1 Execution Model

• Driver: Plans execution, constructs DAGs, manages SparkSession, coordinates tasks, and collects metrics.
• Executors: Worker processes executing task threads, maintaining in-memory storage (RAM), and handling spill to disk.
• Transformations vs Actions:
  • Narrow Transformations: Operations where each input partition maps to at most one output partition (map, filter, flatmap). No network shuffle.
  • Wide Transformations: Operations requiring data exchange across cluster partitions (groupByKey, reduceByKey, join, distinct, repartition). Triggers a Shuffle Stage.
  • Actions: Trigger execution of physical execution plan (count, collect, write, first, take).

1.2 Data Lake storage Formats: Delta Lake & Apache Iceberg

• Acid Transactions: Transaction log (_delta_log or Iceberg metadata trees) providing ACID semantics, time travel, and rollback.
• Medallion Architecture:
  • Bronze (Raw): Append-only ingestion layer preserving original schema (JSON, CSV, Parquet).
  • Silver (Cleaned/Enriched): Deduplicated, typed, normalized, and joined domain entities.
  • Gold (Aggregated/Business): Star-schema dimensions/facts, aggregated metrics, wide tables optimized for BI/ML queries.

---

2. Idempotency & Partitioning Strategies

2.1 Dynamic Partition Overwrite & Upserts

Idempotence guarantees that re-running a pipeline execution produces identical state without duplicate records or partial writes.

PySpark Dynamic Partition Overwrite

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("IdempotentPartitionWriter") \
    .config("spark.sql.sources.partitionOverwriteMode", "dynamic") \
    .getOrCreate()

# Ensures only affected partitions are replaced upon re-run
df_processed.write \
    .format("parquet") \
    .mode("overwrite") \
    .partitionBy("event_date") \
    .save("s3a://data-warehouse/silver/user_events/")

Delta Lake MERGE (Upsert) Pattern

from delta.tables import DeltaTable

def upsert_to_delta(micro_batch_df, batch_id):
    delta_table = DeltaTable.forPath(spark, "s3a://data-warehouse/silver/user_profiles")
    
    delta_table.alias("target").merge(
        micro_batch_df.alias("updates"),
        "target.user_id = updates.user_id"
    ).whenMatchedUpdate(set={
        "email": "updates.email",
        "updated_at": "updates.updated_at",
        "status": "updates.status"
    }).whenNotMatchedInsert(values={
        "user_id": "updates.user_id",
        "email": "updates.email",
        "created_at": "updates.created_at",
        "updated_at": "updates.updated_at",
        "status": "updates.status"
    }).execute()

---

3. Performance Optimization & Tuning

3.1 Skew Mitigation Techniques

Data skew occurs when one partition receives disproportionately more data, stalling stage completion (99% task completion syndrome).

Salting Technique for Skewed Joins

from pyspark.sql import functions as F

SALT_FACTOR = 8

# Add random salt to skewed left table
df_skewed = df_large.withColumn("salt", (F.rand() * SALT_FACTOR).cast("integer"))

# Explode right table to match all salt keys
df_small_salted = df_small.withColumn("salt_array", F.array([F.lit(i) for i in range(SALT_FACTOR)])) \
    .withColumn("salt", F.explode("salt_array")) \
    .drop("salt_array")

# Perform join on join_key AND salt key
df_joined = df_skewed.join(
    df_small_salted,
    on=["join_key", "salt"],
    how="inner"
).drop("salt")

3.2 Adaptive Query Execution (AQE)

Enable AQE to dynamically re-optimize execution plans based on runtime statistics:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.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", "256MB")

3.3 Broadcast Join Tuning

Avoid network shuffles for small lookup tables (<10MB default, up to 1GB with tuning):

from pyspark.sql.functions import broadcast

# Explicitly hint Spark to broadcast small dimension
df_fact.join(broadcast(df_dim_small), "dim_id", "inner")

---

4. Production Anti-Patterns

Anti-Pattern	Operational Risk	Production Remedy
Calling .collect() on large DataFrames	Driver Out-Of-Memory (OOM) crash	Use .take(n), write directly to storage, or sample data
Using DataFrame.toPandas() in distributed loop	Serialization bottleneck & Driver OOM	Use PySpark Pandas API or PyArrow UDFs (pandas_udf)
Unpartitioned .repartition(1) before write	Single executor bottleneck & disk spill	Use .coalesce(n) or dynamic partition overwrites
Row-by-row .iterrows() in Python UDFs	Context switching overhead (JVM to Python)	Native Spark SQL functions or vectorized pandas_udf
Over-partitioning (tiny files problem)	NameNode / S3 metadata strain, slow listing	Target ~128MB - 512MB file sizes, run OPTIMIZE / VORDER

---

5. Production Implementation Blueprint

Scalable Pipeline with Vectorized UDFs & Structural Monitoring

import sys
import logging
from typing import Iterator
import pandas as pd
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType, StructType, StructField, StringType, TimestampType

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("SparkAnalyticsEngine")

def build_spark_session(app_name: str) -> SparkSession:
    """Configures high-throughput Spark session optimized for S3/Delta processing."""
    return SparkSession.builder \
        .appName(app_name) \
        .config("spark.sql.adaptive.enabled", "true") \
        .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
        .getOrCreate()

# Vectorized Pandas UDF for scalable ML/Analytical scoring
@F.pandas_udf(DoubleType())
def calculate_risk_score_udf(amount_series: pd.Series, velocity_series: pd.Series) -> pd.Series:
    """Vectorized calculation executed in Arrow batches across worker nodes."""
    return (amount_series * 0.45) + (velocity_series * )

 () -> :
    
    logger.info()

    raw_df = spark.read.().load(bronze_path) \
        .(F.col() == partition_date)

    
    cleaned_df = raw_df \
        .(F.col().isNotNull()) \
        .dropDuplicates([]) \
        .withColumn(, F.col().cast(DoubleType())) \
        .withColumn(, F.col().cast(DoubleType())) \
        .withColumn(, calculate_risk_score_udf(F.col(), F.col())) \
        .withColumn(, F.current_timestamp())

    logger.info()
    
    cleaned_df.write \
        .() \
        .mode() \
        .option(, ) \
        .save(silver_path)

    logger.info()

 __name__ == :
    spark = build_spark_session()
    :
        process_transaction_bronze_to_silver(
            spark=spark,
            bronze_path=,
            silver_path=,
            partition_date=
        )
     Exception  e:
        logger.error()
        sys.exit()
    :
        spark.stop()