Production Hardening for PySpark Applications

Architecture Diagram: Production Hardening Overview

Architecture Diagram

╔══════════════════════════════════════════════════════════════════════════════════════════╗
║                    PYSPARK PRODUCTION HARDENING FRAMEWORK                                ║
╠══════════════════════════════════════════════════════════════════════════════════════════╣
║                                                                                          ║
║  ┌────────────────────────────────────────────────────────────────────────────────────┐  ║
║  │                    PRODUCTION READINESS PILLARS                                    │  ║
║  │                                                                                    │  ║
║  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐            │  ║
║  │  │ SECURITY │  │GOVERNANCE│  │  CI/CD   │  │MONITORING│  │ TESTING  │            │  ║
║  │  │          │  │          │  │          │  │          │  │          │            │  ║
║  │  │• AuthN/  │  │• Data    │  │• Build   │  │• Metrics │  │• Unit    │            │  ║
║  │  │  AuthZ   │  │  Lineage │  │  Automate│  │  Collect │  │  Tests   │            │  ║
║  │  │• Encrypt │  │• Schema  │  │• Test    │  │• Alert   │  │• Integr. │            │  ║
║  │  │• Secrets │  │  Registry│  │  Gate    │  │  Rules   │  │  Tests   │            │  ║
║  │  │• Audit   │  │• Quality │  │• Deploy  │  │• Dashbrd │  │• Perf    │            │  ║
║  │  │• Network │  │  Gates   │  │  Auto    │  │• Log Agg │  │  Tests   │            │  ║
║  │  └──────────┘  └──────────┘  └──────────┘  └──────────┘  └──────────┘            │  ║
║  └────────────────────────────────────────────────────────────────────────────────────┘  ║
║                                                                                          ║
║  ┌────────────────────────────────────────────────────────────────────────────────────┐  ║
║  │                    PRODUCTION DEPLOYMENT PIPELINE                                   │  ║
║  │                                                                                    │  ║
║  │  ┌────────┐    ┌────────┐    ┌────────┐    ┌────────┐    ┌────────┐    ┌────────┐│  ║
║  │  │  Code  │───▶│  Build │───▶│  Test  │───▶│Security│───▶│ Deploy │───▶│Monitor ││  ║
║  │  │  Push  │    │  & Lint │    │  Suite │    │  Scan  │    │  Stage │    │& Alert ││  ║
║  │  └────────┘    └────────┘    └────────┘    └────────┘    └────────┘    └────────┘│  ║
║  │                                                                                    │  ║
║  │  Trigger:        Duration:       Duration:     Duration:    Duration:    Duration:│  ║
║  │  git push        2-5 min         5-15 min      3-5 min      5-10 min    Continuous│  ║
║  │                                                                                    │  ║
║  │  ┌──────────────────────────────────────────────────────────────────────────────┐ │  ║
║  │  │  Quality Gates (Must Pass Before Promotion)                                   │ │  ║
║  │  │                                                                               │ │  ║
║  │  │  ✓ Code coverage > 80%                                                        │ │  ║
║  │  │  ✓ Zero critical/high security vulnerabilities                                │ │  ║
║  │  │  ✓ Schema validation passed                                                   │ │  ║
║  │  │  ✓ Performance benchmarks met                                                 │ │  ║
║  │  │  ✓ Data quality checks passed                                                 │ │  ║
║  │  │  ✓ Peer review approved                                                       │ │  ║
║  │  └──────────────────────────────────────────────────────────────────────────────┘ │  ║
║  └────────────────────────────────────────────────────────────────────────────────────┘  ║
║                                                                                          ║
╚══════════════════════════════════════════════════════════════════════════════════════════╝

Architecture Diagram: Security Architecture

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────────────────┐
│                    PYSPARK SECURITY ARCHITECTURE                                      │
├─────────────────────────────────────────────────────────────────────────────────────┤
│                                                                                     │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
│  │                    LAYER 1: NETWORK SECURITY                                 │   │
│  │                                                                             │   │
│  │  ┌───────────────────────────────────────────────────────────────────────┐ │   │
│  │  │                                                                       │ │   │
║  │  │  VPC / VNet                                                           │ │   │
║  │  │  ┌─────────────────────────────────────────────────────────────────┐ │ │   │
║  │  │  │  Private Subnet (Spark Cluster)                                 │ │ │   │
║  │  │  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │ │ │   │
║  │  │  │  │ Driver   │  │Executor 1│  │Executor 2│  │Executor N│       │ │ │   │
║  │  │  │  │ Node     │  │          │  │          │  │          │       │ │ │   │
║  │  │  │  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │ │ │   │
║  │  │  │                                                                 │ │ │   │
║  │  │  │  Security Groups:                                              │ │ │   │
║  │  │  │  • Inbound: Only from bastion/Jump host (SSH)                 │ │ │   │
║  │  │  │  • Inbound: Only from driver (executor communication)         │ │ │   │
║  │  │  │  • Outbound: Only to S3/ADLS (VPC Endpoint)                  │ │ │   │
║  │  │  │  • Outbound: Only to Kafka (VPC Endpoint)                    │ │ │   │
║  │  │  └─────────────────────────────────────────────────────────────────┘ │ │   │
║  │  │                                                                       │ │   │
║  │  │  VPC Endpoints (No Internet Traffic):                                │ │   │
║  │  │  • com.amazonaws.*.s3 → S3 Gateway Endpoint                         │ │   │
║  │  │  • com.amazonaws.*.dynamodb → DynamoDB Endpoint                     │ │   │
║  │  │  • com.amazonaws.*.kafka → Kafka Interface Endpoint                 │ │   │
║  │  │  • com.amazonaws.*.glue → Glue Interface Endpoint                   │ │   │
║  │  │                                                                       │ │   │
║  │  └───────────────────────────────────────────────────────────────────────┘ │   │
║  │                                                                             │   │
║  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                                                                     │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
│  │                    LAYER 2: AUTHENTICATION & AUTHORIZATION                   │   │
│  │                                                                             │   │
║  │  ┌───────────────────────────────────────────────────────────────────────┐ │   │
║  │  │                                                                       │ │   │
║  │  │  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐           │ │   │
║  │  │  │  IAM Roles   │    │  Kerberos    │    │  OAuth 2.0   │           │ │   │
║  │  │  │  (Cloud)     │    │  (On-Prem)   │    │  (Cloud)     │           │ │   │
║  │  │  │              │    │              │    │              │           │ │   │
║  │  │  │  • EMR Role  │    │  • Principal │    │  • Token     │           │ │   │
║  │  │  │  • S3 Access │    │  • Keytab    │    │  • Refresh   │           │ │   │
║  │  │  │  • Glue      │    │  • Ticket    │    │  • Scope     │           │ │   │
║  │  │  └──────────────┘    └──────────────┘    └──────────────┘           │ │   │
║  │  │                                                                       │ │   │
║  │  │  ┌─────────────────────────────────────────────────────────────────┐ │ │   │
║  │  │  │  Unity Catalog (Fine-Grained Access Control)                    │ │ │   │
║  │  │  │                                                                 │ │ │   │
║  │  │  │  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐      │ │ │   │
║  │  │  │  │ Catalog  │  │ Schema   │  │ Table    │  │ Column   │      │ │ │   │
║  │  │  │  │ Level    │  │ Level    │  │ Level    │  │ Level    │      │ │ │   │
║  │  │  │  │          │  │          │  │          │  │          │      │ │ │   │
║  │  │  │  │ USE      │  │ CREATE   │  │ SELECT   │  │ MASKING  │      │ │ │   │
║  │  │  │  │ CATALOG  │  │ SCHEMA   │  │ INSERT   │  │ REDACT   │      │ │ │   │
║  │  │  │  │          │  │          │  │ UPDATE   │  │ HASH     │      │ │ │   │
║  │  │  │  │          │  │          │  │ DELETE   │  │          │      │ │ │   │
║  │  │  │  └──────────┘  └──────────┘  └──────────┘  └──────────┘      │ │ │   │
║  │  │  └─────────────────────────────────────────────────────────────────┘ │ │   │
║  │  └───────────────────────────────────────────────────────────────────────┘ │   │
║  │                                                                             │   │
║  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                                                                     │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
│  │                    LAYER 3: DATA PROTECTION                                  │   │
│  │                                                                             │   │
║  │  ┌───────────────────────────────────────────────────────────────────────┐ │   │
║  │  │                                                                       │ │   │
║  │  │  Encryption at Rest:                                                  │ │   │
║  │  │  • S3 SSE-KMS (AWS-managed keys) or SSE-C (customer keys)          │ │   │
║  │  │  • Delta Lake column-level encryption (upcoming)                     │ │   │
║  │  │  • EBS encryption for local disks                                    │ │   │
║  │  │                                                                       │ │   │
║  │  │  Encryption in Transit:                                               │ │   │
║  │  │  • TLS 1.3 for all Spark communication                              │ │   │
║  │  │  • HTTPS for S3/ADLS access                                          │ │   │
║  │  │  • mTLS for Kafka connections                                        │ │   │
║  │  │                                                                       │ │   │
║  │  │  Secrets Management:                                                  │ │   │
║  │  │  • AWS Secrets Manager / Azure Key Vault / GCP Secret Manager        │ │   │
║  │  │  • Never hardcode credentials in code or configs                     │ │   │
║  │  │  • Rotate credentials automatically                                  │ │   │
║  │  │                                                                       │ │   │
║  │  │  Data Masking:                                                        │ │   │
║  │  │  • Column-level masking rules in Unity Catalog                       │ │   │
║  │  │  • Dynamic masking based on user role                                │ │   │
║  │  │  • PII detection and automatic redaction                             │ │   │
║  │  │                                                                       │ │   │
║  │  └───────────────────────────────────────────────────────────────────────┘ │   │
║  │                                                                             │   │
║  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                                                                     │
└─────────────────────────────────────────────────────────────────────────────────────┘

Architecture Diagram: CI/CD Pipeline for PySpark

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────────────────┐
│                    PYSPARK CI/CD PIPELINE                                              │
├─────────────────────────────────────────────────────────────────────────────────────┤
│                                                                                     │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
│  │  DEVELOPMENT                    STAGING                     PRODUCTION     │   │
│  │  ─────────────                  ────────                    ───────────     │   │
║  │                                                                             │   │
║  │  ┌───────────────────────────────────────────────────────────────────────┐ │   │
║  │  │  PHASE 1: BUILD & VALIDATE (2-5 minutes)                              │ │   │
║  │  │                                                                       │ │   │
║  │  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐  │ │   │
║  │  │  │  Lint   │  │  Type   │  │  Unit   │  │  Schema │  │  Build  │  │ │   │
║  │  │  │  Check  │  │  Check  │  │  Tests  │  │  Valid  │  │  Wheel  │  │ │   │
║  │  │  │         │  │         │  │         │  │         │  │         │  │ │   │
║  │  │  │  flake8 │  │  mypy   │  │  pytest │  │  Great  │  │  pip    │  │ │   │
║  │  │  │  black  │  │  pyspark│  │  80%+   │  │  Expect │  │  wheel  │  │ │   │
║  │  │  └─────────┘  └─────────┘  └─────────┘  └─────────┘  └─────────┘  │ │   │
║  │  └───────────────────────────────────────────────────────────────────────┘ │   │
║  │                              │                                              │   │
║  │                              ▼                                              │   │
║  │  ┌───────────────────────────────────────────────────────────────────────┐ │   │
║  │  │  PHASE 2: SECURITY SCAN (3-5 minutes)                                │ │   │
║  │  │                                                                       │ │   │
║  │  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐                │ │   │
║  │  │  │  SAST   │  │  SCA    │  │  Secret │  │  Infra  │                │ │   │
║  │  │  │  Scan   │  │  Scan   │  │  Scan   │  │  Scan   │                │ │   │
║  │  │  │         │  │         │  │         │  │         │                │ │   │
║  │  │  │ bandit  │  │ safety  │  │ truffleH│  │  tfsec  │                │ │   │
║  │  │  │ sonar   │  │ pip     │  │ git     │  │  cfn    │                │ │   │
║  │  │  └─────────┘  └─────────┘  └─────────┘  └─────────┘                │ │   │
║  │  └───────────────────────────────────────────────────────────────────────┘ │   │
║  │                              │                                              │   │
║  │                              ▼                                              │   │
║  │  ┌───────────────────────────────────────────────────────────────────────┐ │   │
║  │  │  PHASE 3: DEPLOY TO STAGING (5-10 minutes)                           │ │   │
║  │  │                                                                       │ │   │
║  │  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐                │ │   │
║  │  │  │  Deploy │  │  Smoke  │  │  Integr │  │  Perf   │                │ │   │
║  │  │  │  to     │  │  Tests  │  │  Tests  │  │  Tests  │                │ │   │
║  │  │  │  Staging│  │         │  │         │  │         │                │ │   │
║  │  │  │         │  │  Basic  │  │  E2E    │  │  Load   │                │ │   │
║  │  │  │  EMR/   │  │  queries│  │  data   │  │  1000   │                │ │   │
║  │  │  │  DBricks│  │  pass   │  │  flows  │  │  TPS    │                │ │   │
║  │  │  └─────────┘  └─────────┘  └─────────┘  └─────────┘                │ │   │
║  │  └───────────────────────────────────────────────────────────────────────┘ │   │
║  │                              │                                              │   │
║  │                              ▼                                              │   │
║  │  ┌───────────────────────────────────────────────────────────────────────┐ │   │
║  │  │  PHASE 4: DEPLOY TO PRODUCTION (5-10 minutes)                        │ │   │
║  │  │                                                                       │ │   │
║  │  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐                │ │   │
║  │  │  │  Blue/  │  │  Canary │  │  Auto   │  │  Roll-  │                │ │   │
║  │  │  │  Green  │  │  Deploy │  │  Roll-  │  │  back   │                │ │   │
║  │  │  │  Deploy │  │  (5%→25%│  │  back   │  │  Ready  │                │ │   │
║  │  │  │         │  │  →100%) │  │  on     │  │         │                │ │   │
║  │  │  │  Zero   │  │         │  │  error  │  │  1-click│                │ │   │
║  │  │  │  down   │  │  Auto   │  │  rate   │  │  revert │                │ │   │
║  │  │  │  time   │  │  metrics│  │  > 1%   │  │         │                │ │   │
║  │  │  └─────────┘  └─────────┘  └─────────┘  └─────────┘                │ │   │
║  │  └───────────────────────────────────────────────────────────────────────┘ │   │
║  │                                                                             │   │
║  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                                                                     │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
║  │  MONITORING & OBSERVABILITY (Continuous)                                     │   │
║  │                                                                             │   │
║  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐         │   │
║  │  │Metrics  │  │ Logs    │  │ Traces  │  │ Alerts  │  │Dashbrd  │         │   │
║  │  │Prometheus│ │ ELK     │  │ Jaeger  │  │PagerDuty│  │Grafana  │         │   │
║  │  │CloudWatch│ │Splunk   │  │ X-Ray   │  │Slack    │  │Custom   │         │   │
║  │  └─────────┘  └─────────┘  └─────────┘  └─────────┘  └─────────┘         │   │
║  │                                                                             │   │
║  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                                                                     │
└─────────────────────────────────────────────────────────────────────────────────────┘

Detailed Explanation

Production hardening transforms a working PySpark prototype into a reliable, secure, and maintainable production system. This requires systematic attention to security, governance, CI/CD automation, monitoring, and testing. The cost of production incidents (data corruption, security breaches, SLA violations) far exceeds the investment in hardening.

Security is the foundation of production readiness. The defense-in-depth approach implements multiple security layers: network security (VPC isolation, security groups, VPC endpoints), authentication and authorization (IAM roles, Kerberos, OAuth 2.0, Unity Catalog), data protection (encryption at rest and in transit, secrets management, column-level masking), and audit logging (all data access and mutations logged). Every PySpark application must authenticate before accessing any resource and be authorized for the specific operations it performs.

CI/CD automation ensures consistent, repeatable deployments with quality gates that prevent regressions. The pipeline validates code quality (linting, type checking), runs unit and integration tests, performs security scans (SAST, SCA, secret detection), deploys to staging for smoke testing, and promotes to production with blue/green or canary deployment strategies. Quality gates block promotion if any check fails, preventing defective code from reaching production.

Monitoring and observability provide visibility into production behavior. Metrics collection tracks key indicators: job duration, data freshness, error rates, resource utilization, and cost. Alerting rules trigger notifications when metrics exceed thresholds. Dashboards provide real-time visibility into system health. Log aggregation enables debugging and forensics. Distributed tracing tracks requests across distributed components.

Testing in PySpark requires specialized approaches due to the distributed nature of Spark. Unit tests use local Spark sessions with small datasets. Integration tests verify end-to-end data flows. Performance tests measure throughput and latency under load. Data quality tests validate schema, completeness, accuracy, and consistency. Contract tests verify compatibility with downstream consumers.

Governance ensures compliance with regulatory requirements (GDPR, HIPAA, SOX) and organizational policies (data classification, retention, access control). Data lineage tracks the flow of data from source to consumption. Schema registries prevent breaking changes. Quality gates enforce data standards. Access policies control who can read and write which data.

Key Concepts Table

Mathematical Foundations

Definition: SLA/SLO

A Service Level Agreement (SLA) defines uptime guarantees, while Service Level Objectives (SLOs) set internal targets:

\text{Availability} = \frac{\text{Uptime}}{\text{Total}} = 1 - \frac{\text{Downtime}}{\text{Total}}

Three nines (99.9%) = 8.76 hours/year downtime; four nines (99.99%) = 52.56 minutes/year.

Error Budget

Remaining error budget for period $T$ with SLO target $s$ :

B(T) = (1 - s) \times T - \text{Downtime}(T)

Deployment is blocked when $B(T) \leq 0$ .

Reliability Theorem

For $n$ independent components with individual reliability $r$ :

R_{\text{system}} = r^n

Series reliability decreases multiplicatively. Parallel redundancy: $R_{\text{parallel}} = 1 - (1-r)^k$ for $k$ redundant components.

Throughput Capacity

Peak throughput with $n$ executors and per-executor throughput $t$ :

T_{\text{peak}} = n \times t \times \text{efficiency}

where efficiency $\in [0.6, 0.9]$ depends on shuffle overhead and skew.

Monitoring Alert Threshold

Alert threshold for metric $M$ with rolling mean $\mu$ and standard deviation $\sigma$ :

\text{Alert} \iff M > \mu + k\sigma

where $k = 3$ for 99.7% confidence (3-sigma rule).

Key Insight

Production hardening is about reducing blast radius: circuit breakers prevent cascading failures, retries with exponential backoff handle transient failures, and health checks enable automatic recovery. The goal is graceful degradation, not perfection.

Summary

Production hardening combines reliability engineering (redundancy, circuit breakers) with operational excellence (monitoring, alerting). SLOs define reliability targets; error budgets govern deployment velocity. System reliability decreases multiplicatively with component count, making redundancy critical.

Key Concepts Table (cont.)

Pillar	Component	Tool/Technology	Implementation	Priority
Security	Network Isolation	VPC, Security Groups	Private subnets, VPC endpoints	Critical
Security	Authentication	IAM, Kerberos, OAuth	Role-based access	Critical
Security	Authorization	Unity Catalog, ACLs	Column-level masking	Critical
Security	Encryption	AWS KMS, TLS 1.3	At rest + in transit	Critical
Security	Secrets	AWS Secrets Manager	Never hardcode	Critical
Governance	Data Lineage	OpenLineage, Unity	Auto-tracked	High
Governance	Schema Registry	Delta Lake, Unity	Schema evolution control	High
Governance	Quality Gates	Great Expectations, Deequ	Automated validation	High
CI/CD	Code Quality	flake8, black, mypy	Pre-commit hooks	High
CI/CD	Testing	pytest, chispa	Unit + integration	High
CI/CD	Security Scanning	bandit, safety, truffleHog	Automated scans	High
CI/CD	Deployment	Blue/Green, Canary	Zero-downtime	Medium
Monitoring	Metrics	Prometheus, CloudWatch	Custom Spark metrics	High
Monitoring	Logging	ELK, Splunk	Structured logging	High
Monitoring	Alerting	PagerDuty, Slack	Threshold-based	High
Monitoring	Dashboards	Grafana, Custom	Real-time visibility	Medium

Code Examples

Example 1: Secure PySpark Application with IAM and Column Masking

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# ─── Secure Spark Session Configuration ───
spark = (
    SparkSession.builder
    .appName("Secure-PySpark-App")
    # Kerberos authentication (on-prem)
    .config("spark.kerberos.keytab", "/etc/security/keytabs/spark.keytab")
    .config("spark.kerberos.principal", "spark/_HOST@COMPANY.COM")
    # Unity Catalog (Databricks)
    .config("spark.databricks.unityCatalog.enabled", "true")
    # Encryption
    .config("spark.ssl.enabled", "true")
    .config("spark.ssl.keyStore", "/path/to/keystore.jks")
    .config("spark.ssl.trustStore", "/path/to/truststore.jks")
    # Enable AQE
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)

# ─── Secrets Management (Never hardcode!) ───
def get_secret(secret_name):
    """Retrieve secret from cloud secrets manager."""
    import boto3  # or azure.keyvault.secrets / google.cloud.secretmanager
    
    client = boto3.client('secretsmanager')
    response = client.get_secret_value(SecretId=secret_name)
    return response['SecretString']

# Use secrets for database connections
db_password = get_secret("prod/warehouse/password")
db_username = get_secret("prod/warehouse/username")

# ─── Data Access with Unity Catalog Policies ───
# Read with row-level security (automatically applied by Unity Catalog)
customer_data = (
    spark.read
    .format("delta")
    .table("sales.customers")  # Unity Catalog table with policies
)

# Column-level masking is applied automatically based on user role
# - Analysts see: customer_id, masked_name, masked_email
# - Data engineers see: customer_id, full_name, email
# - Executives see: all columns

# ─── Data Masking Functions (Custom Implementation) ───
def mask_pii_columns(df, mask_rules):
    """Apply data masking based on column-level rules."""
    masked_df = df
    
    for col_name, rule in mask_rules.items():
        if col_name in df.columns:
            if rule["type"] == "partial_mask":
                # Show first 2 chars + ***
                masked_df = masked_df.withColumn(
                    col_name,
                    when(
                        col(col_name).isNotNull(),
                        concat(
                            substring(col(col_name), 1, 2),
                            lit("***")
                        )
                    ).otherwise(lit(None))
                )
            elif rule["type"] == "full_mask":
                masked_df = masked_df.withColumn(col_name, lit("***"))
            elif rule["type"] == "hash":
                masked_df = masked_df.withColumn(
                    col_name,
                    sha2(col(col_name), 256)
                )
            elif rule["type"] == "redact":
                masked_df = masked_df.withColumn(
                    col_name,
                    regexp_replace(col(col_name), r'.', '*')
                )
    
    return masked_df

# Apply masking for analyst access
mask_rules = {
    "customer_name": {"type": "partial_mask"},
    "email": {"type": "partial_mask"},
    "credit_card": {"type": "full_mask"},
    "ssn": {"type": "hash"},
    "address": {"type": "redact"},
}

masked_data = mask_pii_columns(customer_data, mask_rules)
masked_data.show(5)
# +-----------+--------------+------------------+----------+-------+
# |customer_id|customer_name |             email|credit_card|    ssn|
# +-----------+--------------+------------------+----------+-------+
# |        101| Al***        | al***            |  ***     | a1b2..|
# |        102| Bo***        | bo***            |  ***     | c3d4..|
# +-----------+--------------+------------------+----------+-------+

# ─── Audit Logging ───
def log_data_access(user, table, operation, row_count):
    """Log data access for audit compliance."""
    import boto3
    from datetime import datetime
    
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "user": user,
        "table": table,
        "operation": operation,
        "row_count": row_count,
        "session_id": spark.sparkContext.applicationId,
    }
    
    # Write to CloudWatch Logs
    import json
    print(json.dumps(log_entry))
    
    # Also write to audit table
    spark.createDataFrame([log_entry]).write \
        .format("delta") \
        .mode("append") \
        .save("/mnt/audit/data_access")

log_data_access(
    user="analyst@company.com",
    table="sales.customers",
    operation="SELECT",
    row_count=customer_data.count()
)

Example 2: CI/CD Pipeline Configuration (GitHub Actions)

# .github/workflows/pyspark-ci-cd.yml
name: PySpark CI/CD Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

env:
  PYTHON_VERSION: '3.11'
  SPARK_VERSION: '3.5.0'
  DELTA_VERSION: '3.1.0'

jobs:
  # ─── PHASE 1: Build & Validate ───
  build-and-validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
      
      - name: Install dependencies
        run: |
          pip install -r requirements-dev.txt
          pip install pyspark==${{ env.SPARK_VERSION }} \
                      delta-spark==${{ env.DELTA_VERSION }} \
                      great-expectations chispa
      
      - name: Lint check
        run: |
          flake8 src/ tests/
          black --check src/ tests/
          isort --check-only src/ tests/
      
      - name: Type check
        run: |
          mypy src/ --ignore-missing-imports
      
      - name: Unit tests
        run: |
          pytest tests/unit/ \
            --cov=src/ \
            --cov-report=xml \
            --cov-report=html \
            --junitxml=reports/unit-tests.xml \
            -v
        env:
          PYSPARK_PYTHON: python
      
      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml

  # ─── PHASE 2: Security Scan ───
  security-scan:
    runs-on: ubuntu-latest
    needs: build-and-validate
    steps:
      - uses: actions/checkout@v4
      
      - name: Static Application Security Testing (SAST)
        run: |
          pip install bandit
          bandit -r src/ -f json -o reports/sast-report.json || true
          bandit -r src/ -ll  # Fail on high-severity issues
      
      - name: Software Composition Analysis (SCA)
        run: |
          pip install safety
          safety check --json --output reports/sca-report.json
      
      - name: Secret Scanning
        run: |
          pip install detect-secrets
          detect-secrets scan --all-files --report-json reports/secrets-report.json
      
      - name: Infrastructure Security
        run: |
          pip install checkov
          checkov -d infra/ --framework cloudformation --output junitxml \
            > reports/infra-report.xml

  # ─── PHASE 3: Deploy to Staging ───
  deploy-staging:
    runs-on: ubuntu-latest
    needs: [build-and-validate, security-scan]
    if: github.ref == 'refs/heads/main'
    environment: staging
    steps:
      - uses: actions/checkout@v4
      
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::ACCOUNT:role/deploystaging
          aws-region: us-east-1
      
      - name: Build wheel
        run: |
          pip install build
          python -m build --wheel
      
      - name: Deploy to EMR/Staging
        run: |
          aws emr add-steps \
            --cluster-id ${{ secrets.STAGING_CLUSTER_ID }} \
            --steps Type=Spark,Name=pyspark-app,ActionOnFailure=CONTINUE,Args=[\
              --deploy-mode,cluster,\
              --conf,spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension,\
              s3://staging-bucket/wheels/pyspark-app-${{ github.sha }}.whl]
      
      - name: Smoke tests
        run: |
          pytest tests/smoke/ -v --tb=short
        env:
          STAGING_ENDPOINT: ${{ secrets.STAGING_ENDPOINT }}
      
      - name: Integration tests
        run: |
          pytest tests/integration/ -v --tb=short -m "staging"
        env:
          STAGING_ENDPOINT: ${{ secrets.STAGING_ENDPOINT }}
      
      - name: Performance tests
        run: |
          python tests/performance/benchmark.py --environment staging

  # ─── PHASE 4: Deploy to Production ───
  deploy-production:
    runs-on: ubuntu-latest
    needs: deploy-staging
    if: github.ref == 'refs/heads/main'
    environment: production
    steps:
      - uses: actions/checkout@v4
      
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::ACCOUNT:role/deployproduction
          aws-region: us-east-1
      
      - name: Canary deployment (5% traffic)
        run: |
          # Deploy canary with 5% traffic
          aws emr add-steps \
            --cluster-id ${{ secrets.PROD_CLUSTER_ID }} \
            --steps Type=Spark,Name=pyspark-canary,ActionOnFailure=CONTINUE,Args=[\
              --deploy-mode,cluster,\
              --conf,spark.dynamicAllocation.maxExecutors=2,\
              s3://prod-bucket/wheels/pyspark-app-${{ github.sha }}.whl]
      
      - name: Monitor canary (5 minutes)
        run: |
          sleep 300
          python scripts/monitor_canary.py --duration 300 --error-threshold 0.01
      
      - name: Full deployment
        run: |
          aws emr add-steps \
            --cluster-id ${{ secrets.PROD_CLUSTER_ID }} \
            --steps Type=Spark,Name=pyspark-full,ActionOnFailure=CONTINUE,Args=[\
              --deploy-mode,cluster,\
              --conf,spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension,\
              s3://prod-bucket/wheels/pyspark-app-${{ github.sha }}.whl]
      
      - name: Post-deployment validation
        run: |
          pytest tests/validation/ -v --tb=short

Example 3: Production Monitoring and Alerting

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import logging

# ─── Custom Metrics Collection ───
class PySparkMetrics:
    """Custom metrics for PySpark production monitoring."""
    
    def __init__(self, app_name):
        self.app_name = app_name
        
        # Prometheus metrics
        self.job_duration = Histogram(
            'pyspark_job_duration_seconds',
            'Duration of PySpark jobs in seconds',
            ['job_name', 'status'],
            buckets=[10, 30, 60, 120, 300, 600, 1800, 3600]
        )
        self.records_processed = Counter(
            'pyspark_records_processed_total',
            'Total records processed',
            ['job_name', 'source', 'sink']
        )
        self.records_failed = Counter(
            'pyspark_records_failed_total',
            'Total records that failed processing',
            ['job_name', 'error_type']
        )
        self.data_freshness = Gauge(
            'pyspark_data_freshness_seconds',
            'Seconds since last data update',
            ['table_name']
        )
        self.executor_count = Gauge(
            'pyspark_executor_count',
            'Number of active executors',
            ['cluster_id']
        )
        self.shuffle_bytes = Counter(
            'pyspark_shuffle_bytes_total',
            'Total shuffle bytes',
            ['job_name']
        )
        
        # Start Prometheus metrics server
        start_http_server(8080)
    
    def track_job(self, job_name):
        """Decorator to track job execution metrics."""
        def decorator(func):
            def wrapper(*args, **kwargs):
                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    duration = time.time() - start_time
                    self.job_duration.labels(
                        job_name=job_name, status='success'
                    ).observe(duration)
                    logging.info(f"Job {job_name} completed in {duration:.2f}s")
                    return result
                except Exception as e:
                    duration = time.time() - start_time
                    self.job_duration.labels(
                        job_name=job_name, status='failure'
                    ).observe(duration)
                    self.records_failed.labels(
                        job_name=job_name, error_type=type(e).__name__
                    ).inc()
                    logging.error(f"Job {job_name} failed: {e}")
                    raise
            return wrapper
        return decorator

# ─── Data Quality Monitoring ───
class DataQualityMonitor:
    """Monitor data quality metrics in production."""
    
    def __init__(self, spark):
        self.spark = spark
    
    def check_freshness(self, table_path, max_age_hours=24):
        """Check if data is fresh enough."""
        df = self.spark.read.format("delta").load(table_path)
        
        # Get latest record timestamp
        latest = df.agg(
            max("_updated_at").alias("latest_timestamp")
        ).first()["latest_timestamp"]
        
        if latest is None:
            return {"status": "ERROR", "message": "No data found"}
        
        age_hours = (datetime.now() - latest).total_seconds() / 3600
        
        if age_hours > max_age_hours:
            return {
                "status": "ALERT",
                "message": f"Data is {age_hours:.1f} hours old (max: {max_age_hours})",
                "age_hours": age_hours
            }
        
        return {"status": "OK", "age_hours": age_hours}
    
    def check_completeness(self, df, required_columns):
        """Check completeness of required columns."""
        total_rows = df.count()
        results = {}
        
        for col_name in required_columns:
            null_count = df.filter(col(col_name).isNull()).count()
            completeness = 1 - (null_count / total_rows) if total_rows > 0 else 0
            
            results[col_name] = {
                "completeness": completeness,
                "null_count": null_count,
                "status": "OK" if completeness >= 0.99 else "ALERT"
            }
        
        return results
    
    def check_uniqueness(self, df, key_columns):
        """Check uniqueness of key columns."""
        total_rows = df.count()
        unique_rows = df.select(key_columns).distinct().count()
        
        duplicates = total_rows - unique_rows
        uniqueness = unique_rows / total_rows if total_rows > 0 else 0
        
        return {
            "total_rows": total_rows,
            "unique_rows": unique_rows,
            "duplicates": duplicates,
            "uniqueness": uniqueness,
            "status": "OK" if duplicates == 0 else "ALERT"
        }

# ─── Structured Logging ───
def setup_logging(app_name):
    """Configure structured logging for PySpark."""
    import json
    
    class StructuredFormatter(logging.Formatter):
        def format(self, record):
            log_entry = {
                "timestamp": self.formatTime(record),
                "level": record.levelname,
                "app": app_name,
                "message": record.getMessage(),
                "module": record.module,
                "function": record.funcName,
                "line": record.lineno,
            }
            
            if hasattr(record, 'extra_data'):
                log_entry.update(record.extra_data)
            
            return json.dumps(log_entry)
    
    handler = logging.StreamHandler()
    handler.setFormatter(StructuredFormatter())
    
    logger = logging.getLogger(app_name)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    
    return logger

# ─── Example Usage ───
spark = SparkSession.builder \
    .appName("Production-Monitoring") \
    .getOrCreate()

metrics = PySparkMetrics("production-app")
quality_monitor = DataQualityMonitor(spark)
logger = setup_logging("production-app")

# Track a job
@metrics.track_job("process_transactions")
def process_transactions():
    """Process transaction data with monitoring."""
    df = spark.read.format("delta").load("/mnt/lakehouse/silver/transactions")
    
    # Quality checks
    freshness = quality_monitor.check_freshness(
        "/mnt/lakehouse/silver/transactions",
        max_age_hours=4
    )
    
    if freshness["status"] == "ALERT":
        logger.warning(f"Data freshness alert: {freshness['message']}")
    
    completeness = quality_monitor.check_completeness(
        df, ["transaction_id", "customer_id", "amount"]
    )
    
    for col_name, result in completeness.items():
        if result["status"] == "ALERT":
            logger.warning(f"Completeness alert for {col_name}: {result['completeness']}")
    
    # Process data
    result = (
        df
        .groupBy("customer_id")
        .agg(count("*").alias("transaction_count"))
        .write
        .format("delta")
        .mode("overwrite")
        .save("/mnt/lakehouse/gold/customer_summary")
    )
    
    # Update metrics
    metrics.records_processed.labels(
        job_name="process_transactions",
        source="silver/transactions",
        sink="gold/customer_summary"
    ).inc(df.count())
    
    return result

# Run with monitoring
process_transactions()

Performance Metrics

Metric	Without Hardening	With Hardening	Improvement
Mean Time to Recovery (MTTR)	4-8 hours	15-30 minutes	90% reduction
Production Incidents/Month	8-12	1-2	85% reduction
Deployment Frequency	Weekly	Daily	7x increase
Lead Time for Changes	2-4 weeks	1-2 days	90% reduction
Change Failure Rate	15-25%	2-5%	80% reduction
Security Vulnerabilities	20+/quarter	0-2/quarter	90% reduction
Data Quality Issues	10+/month	1-2/month	85% reduction
SLA Compliance	95%	99.9%	5% improvement
Cost of Downtime	$50K/hour \|$ 5K/hour	90% reduction
Audit Compliance	Manual (days)	Automated (hours)	95% faster

Best Practices

Implement defense-in-depth security — Never rely on a single security control. Use network isolation, authentication, authorization, encryption, and audit logging together. Each layer provides independent protection.
Automate everything with CI/CD — Manual deployments are error-prone and slow. Automate building, testing, security scanning, and deployment. Use quality gates to prevent defective code from reaching production.
Test at every level — Unit tests validate individual functions, integration tests validate data flows, performance tests validate SLAs, and contract tests validate downstream compatibility. Aim for 80%+ code coverage.
Monitor proactively, not reactively — Set up metrics, alerts, and dashboards before production issues occur. Monitor data freshness, quality, performance, and cost continuously. Alert on anomalies, not just failures.
Use structured logging — Log in JSON format with consistent fields (timestamp, level, app, message, context). This enables efficient log searching, aggregation, and analysis in production.
Implement blue/green or canary deployments — Never deploy directly to production. Use deployment strategies that allow instant rollback if issues are detected. Canary deployments (5% → 25% → 100%) reduce blast radius.
Maintain runbooks for common issues — Document step-by-step procedures for common production issues (job failures, data quality problems, performance degradation). This reduces MTTR and enables junior engineers to handle incidents.
Enforce governance as code — Use automated tools (Unity Catalog, Great Expectations, Deequ) to enforce data governance policies. Manual governance processes don't scale and are prone to human error.
Implement comprehensive audit logging — Log all data access, mutations, and configuration changes. This is required for regulatory compliance (GDPR, HIPAA, SOX) and enables forensic analysis after incidents.
Practice chaos engineering — Regularly inject failures (spot interruptions, network partitions, disk failures) to validate that your production hardening actually works. Use tools like Chaos Monkey or custom fault injection.

See also: Snowflake Time Travel (snowflake/02), Kafka CDC (kafka/04), Airflow DAGs (airflow/02)

Production Hardening for PySpark Applications

Production Hardening for PySpark Applications

Architecture Diagram: Production Hardening Overview

Architecture Diagram: Security Architecture

Architecture Diagram: CI/CD Pipeline for PySpark

Detailed Explanation

Key Concepts Table

Mathematical Foundations

Definition: SLA/SLO

Error Budget

Reliability Theorem

Throughput Capacity

Monitoring Alert Threshold

Key Insight

Summary

Key Concepts Table (cont.)

Code Examples

Example 1: Secure PySpark Application with IAM and Column Masking

Example 2: CI/CD Pipeline Configuration (GitHub Actions)

Example 3: Production Monitoring and Alerting

Performance Metrics

Best Practices

Need Expert PySpark Help?