Pyspark and apache spark analytics

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 * 1.85)

def process_transaction_bronze_to_silver(
    spark: SparkSession,
    bronze_path: str,
    silver_path: str,
    partition_date: str
) -> None:
    """Idempotent Bronze-to-Silver transformation pipeline."""
    logger.info(f"Ingesting Bronze payload from {bronze_path} for date: {partition_date}")

    raw_df = spark.read.format("delta").load(bronze_path) \
        .filter(F.col("event_date") == partition_date)

    # Data Quality Cleanse & Deduplication
    cleaned_df = raw_df \
        .filter(F.col("transaction_id").isNotNull()) \
        .dropDuplicates(["transaction_id"]) \
        .withColumn("amount", F.col("amount").cast(DoubleType())) \
        .withColumn("velocity", F.col("velocity").cast(DoubleType())) \
        .withColumn("risk_score", calculate_risk_score_udf(F.col("amount"), F.col("velocity"))) \
        .withColumn("processed_at", F.current_timestamp())

    logger.info("Writing clean records to Silver Delta table with partition isolation...")
    
    cleaned_df.write \
        .format("delta") \
        .mode("overwrite") \
        .option("replaceWhere", f"event_date = '{partition_date}'") \
        .save(silver_path)

    logger.info(f"Successfully processed Silver partition: {partition_date}")

if __name__ == "__main__":
    spark = build_spark_session("ProductionSparkPipeline")
    try:
        process_transaction_bronze_to_silver(
            spark=spark,
            bronze_path="s3a://enterprise-datalake/bronze/transactions",
            silver_path="s3a://enterprise-datalake/silver/transactions",
            partition_date="2026-07-18"
        )
    except Exception as e:
        logger.error(f"Pipeline execution failure: {str(e)}")
        sys.exit(1)
    finally:
        spark.stop()