20. Monitoring & Metrics in PySpark

DfSpark UI

The Spark UI is a web interface (port 4040 during execution, port 4041 after termination) that provides real-time visibility into job progress, stage details, task metrics, storage usage, and SQL query plans.

DfMetrics System

The metrics system collects and exports Spark internal metrics (CPU, memory, shuffle I/O, GC time) to external monitoring systems like Graphite, Prometheus, or JMX via pluggable sinks.

T_{task} = T_{deserialize} + T_{compute} + T_{serialize} + T_{result\\_send}

Shuffle Read/Write Formula

S_{shuffle} = N_{partitions} \\times S_{partition\\_avg} \\times F_{compression}

Here,

$S_{shuffle}$ =Total shuffle data transferred
$N_{partitions}$ =Number of shuffle partitions
$S_{partition\_avg}$ =Average size per partition
$F_{compression}$ =Compression ratio (typically 0.3–0.7)

Key Spark UI metrics to monitor: Task Duration (identify stragglers), Shuffle Read/Write (bottlenecks), GC Time (memory pressure), and Peak Execution Memory (data skew).

Enable event logging for post-mortem analysis: spark.eventLog.enabled=true. Use Spark History Server (port 18080) to replay completed applications and identify performance regressions.

ThStraggler Detection

Theorem: A task is a straggler if its duration exceeds median_task_duration × spark.speculation.multiplier (default 1.5). Speculative execution launches a backup copy; the first to complete is used. This bounds the tail latency to max(median, straggler_duration).

Spark UI: real-time visibility into jobs, stages, tasks, storage, SQL
Metrics system: export to Prometheus, Graphite, JMX via pluggable sinks
Monitor: task duration, shuffle I/O, GC time, peak execution memory
Enable event logging + History Server for post-mortem analysis
Speculative execution handles stragglers by launching backup copies

🏗️ Monitoring Architecture

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────┐
│                    PYSPARK MONITORING ARCHITECTURE                       │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    SPARK UI LAYERS                               │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Layer 1: APPLICATION LEVEL                             │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Spark UI (port 4040)                           │    │    │   │
│  │  │  │  • Jobs tab: Job progress and status            │    │    │   │
│  │  │  │  • Stages tab: Stage details and tasks          │    │    │   │
│  │  │  │  • Storage tab: Cached data and storage usage   │    │    │   │
│  │  │  │  • Environment tab: Configuration and properties│    │    │   │
│  │  │  │  • Executors tab: Executor status and metrics   │    │    │   │
│  │  │  │  • SQL tab: Query execution plans               │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Layer 2: HISTORY LEVEL                                │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  History Server (port 18080)                     │    │    │   │
│  │  │  │  • Completed application history                 │    │    │   │
│  │  │  │  • Job and stage statistics                      │    │    │   │
│  │  │  │  • Executor metrics over time                    │    │    │   │
│  │  │  │  • Configuration history                         │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Layer 3: CLUSTER LEVEL                                │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Cluster Manager UI                              │    │    │   │
│  │  │  │  • YARN ResourceManager UI (port 8088)          │    │    │   │
│  │  │  │  • Kubernetes Dashboard                          │    │    │   │
│  │  │  │  • Standalone Master UI (port 8080)             │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    METRICS SYSTEMS                               │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Spark Metrics System                                   │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Sources:                                         │    │    │   │
│  │  │  │  • ApplicationSource                             │    │    │   │
│  │  │  │  • JVMSource                                    │    │    │   │
│  │  │  │  • CodeGenerationSource                         │    │    │   │
│  │  │  │  • DAGSchedulerSource                           │    │    │   │
│  │  │  │  • ExecutorAllocationManagerSource              │    │    │   │
│  │  │  │  • JvmMetricSource                              │    │    │   │
│  │  │  │  • JobProgressSource                            │    │    │   │
│  │  │  │  • MapOutputTrackerSource                       │    │    │   │
│  │  │  │  • OperatingSystemSource                        │    │    │   │
│  │  │  │  • PoolSource                                   │    │    │   │
│  │  │  │  • SparkSessionSource                           │    │    │   │
│  │  │  │  • SystemMetricsSource                          │    │    │   │
│  │  │  │  • TaburedStatisticsSource                      │    │    │   │
│  │  │  │  • ThreadAllocationRateSource                   │    │    │   │
│  │  │  │  • WheelerSource                                │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  │                                                         │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Sinks:                                          │    │    │   │
│  │  │  │  • ConsoleSink                                  │    │    │   │
│  │  │  │  • CsvSink                                      │    │    │   │
│  │  │  │  • JmxSink                                      │    │    │   │
│  │  │  │  • GraphiteSink                                 │    │    │   │
│  │  │  │  • StatsdSink                                   │    │    │   │
│  │  │  │  • PrometheusSink                               │    │    │   │
│  │  │  │  • Slf4jSink                                    │    │    │   │
│  │  │  │  • ServletSink                                  │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────┘

📊 Metrics Collection Flow

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────┐
│                    METRICS COLLECTION FLOW                               │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    METRIC SOURCES                                │   │
│  │                                                                   │   │
│  │  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐  │   │
│  │  │  Application    │  │  JVM            │  │  System         │  │   │
│  │  │  Metrics        │  │  Metrics        │  │  Metrics        │  │   │
│  │  │                 │  │                 │  │                 │  │   │
│  │  │  • Job count    │  │  • Heap usage   │  │  • CPU usage    │  │   │
│  │  │  • Stage count  │  │  • GC stats     │  │  • Memory usage │  │   │
│  │  │  • Task count   │  │  • Thread count │  │  • Disk I/O     │  │   │
│  │  │  • Shuffle size │  │  • Buffer pool  │  │  • Network I/O  │  │   │
│  │  └─────────────────┘  └─────────────────┘  └─────────────────┘  │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                              │                                          │
│                              ▼                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    METRIC REGISTRY                               │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  MetricRegistry                                        │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  • Register metrics from sources               │    │    │   │
│  │  │  │  • Manage metric lifecycle                      │    │    │   │
│  │  │  │  • Handle metric updates                        │    │    │   │
│  │  │  │  • Provide metric access API                    │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  │                                                         │    │   │
│  │  │  Metric Types:                                         │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Counter: Incrementing values                   │    │    │   │
│  │  │  │  Gauge: Current values                          │    │    │   │
│  │  │  │  Histogram: Value distributions                 │    │    │   │
│  │  │  │  Meter: Event rates                             │    │    │   │
│  │  │  │  Timer: Duration measurements                   │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                              │                                          │
│                              ▼                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    METRIC SINKS                                  │   │
│  │                                                                   │   │
│  │  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐  │   │
│  │  │  Console Sink   │  │  JMX Sink       │  │  External Sinks │  │   │
│  │  │                 │  │                 │  │                 │  │   │
│  │  │  • Debug output │  │  • JConsole     │  │  • Graphite     │  │   │
│  │  │  • Development  │  │  • Monitoring   │  │  • Prometheus   │  │   │
│  │  │  • Testing      │  │    tools        │  │  • StatsD       │  │   │
│  │  └─────────────────┘  └─────────────────┘  └─────────────────┘  │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                              │                                          │
│                              ▼                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    MONITORING DASHBOARD                          │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐    │    │   │
│  │  │  │  Real-time  │  │  Historical │  │  Alerting   │    │    │   │
│  │  │  │  Dashboard  │  │  Trends     │  │  System     │    │    │   │
│  │  │  │             │  │             │  │             │    │    │   │
│  │  │  │  • Live     │  │  • Long     │  │  • Threshold│    │    │   │
│  │  │  │    metrics  │  │    term     │  │    alerts   │    │    │   │
│  │  │  │  • Current  │  │    trends   │  │  • Anomaly  │    │    │   │
│  │  │  │    status   │  │  • Pattern  │  │    detection│    │    │   │
│  │  │  │             │  │    analysis │  │             │    │    │   │
│  │  │  └─────────────┘  └─────────────┘  └─────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────┘

🔍 Performance Monitoring Dashboard

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────┐
│                    PERFORMANCE MONITORING DASHBOARD                      │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    JOB EXECUTION METRICS                         │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Job Progress:                                          │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Job 1: ████████████████████░░░░░░░░ 75% (Running)│    │    │   │
│  │  │  │  Job 2: ████████████████████████████ 100% (Done) │    │    │   │
│  │  │  │  Job 3: ░░░░░░░░░░░░░░░░░░░░░░░░░░░░ 0% (Pending)│    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  │                                                         │    │   │
│  │  │  Stage Progress:                                        │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Stage 1: Map  ████████████████░░░░ 80%         │    │    │   │
│  │  │  │  Stage 1: Reduce ████░░░░░░░░░░░░░░ 25%         │    │    │   │
│  │  │  │  Stage 2: Map  ░░░░░░░░░░░░░░░░░░░░ 0%          │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    EXECUTOR METRICS                              │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Executor Status:                                        │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Executor 1: [■■■■■■■■] Memory: 75%  Cores: 4/4 │    │    │   │
│  │  │  │  Executor 2: [■■■■■■■■] Memory: 68%  Cores: 3/4 │    │    │   │
│  │  │  │  Executor 3: [■■■■■■■■] Memory: 82%  Cores: 4/4 │    │    │   │
│  │  │  │  Executor 4: [■■■■■■■■] Memory: 45%  Cores: 2/4 │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  │                                                         │    │   │
│  │  │  Aggregate Metrics:                                     │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Total Tasks: 1,250  Completed: 1,000 (80%)    │    │    │   │
│  │  │  │  Shuffle Read: 2.5 GB  Shuffle Write: 1.8 GB   │    │    │   │
│  │  │  │  GC Time: 15s (2.5% of total time)             │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    STORAGE METRICS                               │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────┐    │   │
│  │  │  Cached Data:                                           │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Table: users     │ Size: 2.5 GB │ Partitions: 20│    │    │   │
│  │  │  │  Table: orders    │ Size: 1.8 GB │ Partitions: 15│    │    │   │
│  │  │  │  Table: products  │ Size: 0.5 GB │ Partitions: 5 │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  │                                                         │    │   │
│  │  │  Storage Usage:                                         │    │   │
│  │  │  ┌─────────────────────────────────────────────────┐    │    │   │
│  │  │  │  Total Cached: 4.8 GB  Max Storage: 10 GB      │    │    │   │
│  │  │  │  Memory Fraction: 0.6  Storage Fraction: 0.4   │    │    │   │
│  │  │  └─────────────────────────────────────────────────┘    │    │   │
│  │  └─────────────────────────────────────────────────────────┘    │   │
│  └─────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────┘

📚 Detailed Explanation

Monitoring and metrics are essential for maintaining the health, performance, and reliability of PySpark applications. A comprehensive monitoring strategy provides visibility into application behavior, enables proactive issue detection, and supports capacity planning and optimization.

The Spark UI is the primary interface for monitoring application execution. It provides real-time information about jobs, stages, tasks, executors, and storage. The UI is available during application execution (port 4040) and can be accessed after completion through the History Server (port 18080). Understanding the UI's tabs and metrics is essential for debugging performance issues and optimizing applications.

The Spark Metrics System provides a flexible framework for collecting and reporting application metrics. It supports multiple sources (JVM, application, system metrics) and sinks (console, JMX, external systems). Metrics can be configured through the metrics.properties file or programmatically, allowing organizations to integrate Spark metrics with their existing monitoring infrastructure.

Key performance metrics include job and stage completion times, task execution statistics, shuffle read/write volumes, memory usage, and garbage collection overhead. These metrics help identify bottlenecks, detect resource constraints, and optimize application configurations.

The History Server provides persistent storage and visualization of completed application metrics. It enables historical analysis, trend identification, and comparison of different application runs. The History Server is essential for production environments where application performance needs to be tracked over time.

External monitoring systems like Prometheus, Graphite, and StatsD can integrate with Spark's metrics system to provide enterprise-grade monitoring capabilities. These systems offer advanced features like alerting, dashboards, and long-term data retention, making them suitable for production deployments.

Best practices for monitoring include: enabling event logging for all applications, configuring appropriate metrics sinks, setting up alerting for critical metrics, regularly reviewing performance trends, and maintaining monitoring documentation.

Advanced monitoring techniques include custom metrics sources for application-specific metrics, distributed tracing for complex workflows, and machine learning-based anomaly detection for predictive monitoring. These techniques help organizations move from reactive to proactive monitoring.

📊 Key Concepts Table

Metric Category	Key Metrics	Threshold	Action
Job Performance	Job duration, stage count	> 30min	Optimize code, increase resources
Task Execution	Task duration, failure rate	> 10min, > 5%	Check data skew, increase parallelism
Memory Usage	Heap usage, GC time	> 80%, > 5%	Increase memory, tune GC
Shuffle I/O	Shuffle read/write	> 10GB	Optimize partitions, use broadcast
Storage	Cache hit ratio, storage usage	< 90%, > 80%	Tune caching, increase storage
Cluster Resources	Executor utilization, CPU usage	< 70%, > 90%	Adjust allocation, optimize workload

💻 Code Examples

Basic Monitoring Configuration

from pyspark.sql import SparkSession

# Configure Spark with monitoring
spark = SparkSession.builder \
    .appName("MonitoredApplication") \
    .config("spark.eventLog.enabled", "true") \
    .config("spark.eventLog.dir", "hdfs:///spark-history") \
    .config("spark.history.fs.logDirectory", "hdfs:///spark-history") \
    .config("spark.history.ui.port", "18080") \
    .config("spark.ui.enabled", "true") \
    .config("spark.ui.port", "4040") \
    .config("spark.ui.whitelist.axes", "jvm,system") \
    .config("spark.metrics.conf", "/path/to/metrics.properties") \
    .config("spark.metrics.listeners", "org.apache.spark.metrics.MetricsSystem") \
    .getOrCreate()

# Access Spark UI
print(f"Spark UI available at: http://localhost:4040")
print(f"History Server available at: http://localhost:18080")

# Run application with monitoring
df = spark.range(1000000).repartition(100)
result = df.groupBy(col("id") % 10).count().collect()

print(f"Application completed with {len(result)} groups")

spark.stop()

Custom Metrics Collection

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

spark = SparkSession.builder \
    .appName("CustomMetrics") \
    .getOrCreate()

# Get Spark context for metrics
sc = spark.sparkContext
metrics = sc._jsc.status()

# Collect custom metrics
def collect_custom_metrics():
    """Collect custom application metrics"""
    
    # Get JVM metrics
    jvm = metrics.getJvm()
    memory_pools = jvm.getMemoryPools()
    
    custom_metrics = {}
    
    # Memory metrics
    for pool in memory_pools:
        pool_name = pool.getName()
        used = pool.getMemoryUsed()
        max_size = pool.getMaxSize()
        custom_metrics[f"memory.{pool_name}.used"] = used
        custom_metrics[f"memory.{pool_name}.max"] = max_size
        if max_size > 0:
            custom_metrics[f"memory.{pool_name}.usage"] = used / max_size
    
    # GC metrics
    gc_pools = jvm.getGarbageCollectorPools()
    for gc_pool in gc_pools:
        gc_name = gc_pool.getName()
        gc_count = gc_pool.getCollectionCount()
        gc_time = gc_pool.getCollectionTime()
        custom_metrics[f"gc.{gc_name}.count"] = gc_count
        custom_metrics[f"gc.{gc_name}.time"] = gc_time
    
    # Thread metrics
    thread_count = jvm.getThreadCount()
    peak_thread_count = jvm.getPeakThreadCount()
    custom_metrics["jvm.thread_count"] = thread_count
    custom_metrics["jvm.peak_thread_count"] = peak_thread_count
    
    return custom_metrics

# Monitor application performance
def monitor_performance(df, operation_name):
    """Monitor performance of a DataFrame operation"""
    start_time = time.time()
    
    # Execute operation
    result = df.collect()
    
    end_time = time.time()
    duration = end_time - start_time
    
    # Collect metrics
    custom_metrics = collect_custom_metrics()
    
    # Log performance
    print(f"\nOperation: {operation_name}")
    print(f"Duration: {duration:.2f} seconds")
    print(f"Records processed: {len(result)}")
    print(f"Memory usage: {custom_metrics.get('memory.heap.used', 0) / 1024 / 1024:.2f} MB")
    
    return result, duration, custom_metrics

# Test with different operations
df = spark.range(1000000)

# Monitor count operation
result, duration, metrics = monitor_performance(df, "count")

# Monitor groupBy operation
grouped_df = df.groupBy(col("id") % 100)
result, duration, metrics = monitor_performance(grouped_df, "groupBy")

# Monitor join operation
df2 = spark.range(1000000).withColumn("value", col("id") * 2)
joined_df = df.join(df2, "id")
result, duration, metrics = monitor_performance(joined_df, "join")

spark.stop()

Metrics Export to External System

from pyspark.sql import SparkSession
import json
import time

spark = SparkSession.builder \
    .appName("MetricsExport") \
    .getOrCreate()

class MetricsExporter:
    """Export Spark metrics to external systems"""
    
    def __init__(self, app_name):
        self.app_name = app_name
        self.metrics_history = []
    
    def export_to_console(self, metrics):
        """Export metrics to console"""
        print(f"\n{'='*60}")
        print(f"Application: {self.app_name}")
        print(f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"{'='*60}")
        
        for key, value in metrics.items():
            print(f"{key}: {value}")
        
        print(f"{'='*60}\n")
    
    def export_to_json(self, metrics, filename):
        """Export metrics to JSON file"""
        metrics_data = {
            "app_name": self.app_name,
            "timestamp": time.time(),
            "metrics": metrics
        }
        
        self.metrics_history.append(metrics_data)
        
        with open(filename, 'w') as f:
            json.dump(self.metrics_history, f, indent=2)
        
        print(f"Metrics exported to {filename}")
    
    def export_to_prometheus(self, metrics, pushgateway_url):
        """Export metrics to Prometheus Pushgateway"""
        try:
            import requests
            
            # Format metrics for Prometheus
            prometheus_metrics = []
            for key, value in metrics.items():
                # Convert key to Prometheus format
                prom_key = key.replace(".", "_")
                prometheus_metrics.append(f"{prom_key} {value}")
            
            # Push to Pushgateway
            response = requests.post(
                f"{pushgateway_url}/metrics/job/{self.app_name}",
                data='\n'.join(prometheus_metrics)
            )
            
            if response.status_code == 200:
                print(f"Metrics exported to Prometheus Pushgateway")
            else:
                print(f"Failed to export metrics: {response.status_code}")
                
        except ImportError:
            print("requests library not installed")
        except Exception as e:
            print(f"Error exporting to Prometheus: {e}")

# Create metrics exporter
exporter = MetricsExporter("MySparkApplication")

# Collect and export metrics
def collect_and_export_metrics():
    """Collect metrics and export to various systems"""
    
    # Get Spark context
    sc = spark.sparkContext
    jvm = sc._jsc.status().getJvm()
    
    # Collect metrics
    metrics = {
        "spark.executor.memory.used": jvm.getHeapMemoryUsed(),
        "spark.executor.memory.max": jvm.getHeapMemoryMax(),
        "spark.executor.memory.free": jvm.getHeapMemoryFree(),
        "spark.jvm.thread.count": jvm.getThreadCount(),
        "spark.jvm.uptime": jvm.getUptime(),
    }
    
    # Add GC metrics
    gc_pools = jvm.getGarbageCollectorPools()
    for i, gc_pool in enumerate(gc_pools):
        metrics[f"spark.gc.{i}.count"] = gc_pool.getCollectionCount()
        metrics[f"spark.gc.{i}.time"] = gc_pool.getCollectionTime()
    
    # Export to different systems
    exporter.export_to_console(metrics)
    exporter.export_to_json(metrics, "metrics.json")
    
    return metrics

# Monitor application
df = spark.range(1000000)
result = df.groupBy(col("id") % 10).count().collect()

# Export final metrics
final_metrics = collect_and_export_metrics()

spark.stop()

Alerting System

from pyspark.sql import SparkSession
import time
import smtplib
from email.mime.text import MIMEText

spark = SparkSession.builder \
    .appName("AlertingSystem") \
    .getOrCreate()

class AlertingSystem:
    """Alerting system for Spark applications"""
    
    def __init__(self, email_config=None):
        self.email_config = email_config
        self.alert_history = []
    
    def check_threshold(self, metric_name, value, threshold, alert_type="warning"):
        """Check if metric exceeds threshold"""
        if alert_type == "warning" and value > threshold:
            self.send_alert(metric_name, value, threshold, "WARNING")
            return True
        elif alert_type == "critical" and value > threshold:
            self.send_alert(metric_name, value, threshold, "CRITICAL")
            return True
        return False
    
    def send_alert(self, metric_name, value, threshold, level):
        """Send alert notification"""
        alert_message = f"""
        Spark Application Alert
        
        Level: {level}
        Metric: {metric_name}
        Current Value: {value}
        Threshold: {threshold}
        Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}
        """
        
        print(f"\n{'!'*60}")
        print(f"ALERT [{level}]: {metric_name}")
        print(f"Value: {value} exceeds threshold: {threshold}")
        print(f"{'!'*60}\n")
        
        # Store alert
        self.alert_history.append({
            "level": level,
            "metric": metric_name,
            "value": value,
            "threshold": threshold,
            "timestamp": time.time()
        })
        
        # Send email if configured
        if self.email_config and level == "CRITICAL":
            self.send_email_alert(alert_message)
    
    def send_email_alert(self, message):
        """Send email alert for critical issues"""
        try:
            msg = MIMEText(message)
            msg['Subject'] = 'Spark Application Critical Alert'
            msg['From'] = self.email_config['from']
            msg['To'] = self.email_config['to']
            
            with smtplib.SMTP(self.email_config['smtp_server']) as server:
                server.send_message(msg)
            
            print("Critical alert email sent")
        except Exception as e:
            print(f"Failed to send email alert: {e}")
    
    def get_alert_history(self):
        """Get alert history"""
        return self.alert_history

# Create alerting system
alerting = AlertingSystem()

# Monitor metrics and check thresholds
def monitor_with_alerting():
    """Monitor metrics with alerting"""
    
    sc = spark.sparkContext
    jvm = sc._jsc.status().getJvm()
    
    # Get metrics
    heap_used = jvm.getHeapMemoryUsed()
    heap_max = jvm.getHeapMemoryMax()
    heap_usage = heap_used / heap_max if heap_max > 0 else 0
    
    # Check thresholds
    alerting.check_threshold("heap_usage", heap_usage * 100, 80, "warning")
    alerting.check_threshold("heap_usage", heap_usage * 100, 95, "critical")
    
    # Get GC metrics
    gc_pools = jvm.getGarbageCollectorPools()
    for gc_pool in gc_pools:
        gc_time = gc_pool.getCollectionTime()
        alerting.check_threshold(f"gc_time_{gc_pool.getName()}", gc_time, 5000, "warning")
    
    return {
        "heap_used": heap_used,
        "heap_max": heap_max,
        "heap_usage_percent": heap_usage * 100
    }

# Monitor application
df = spark.range(1000000)
result = df.groupBy(col("id") % 10).count().collect()

# Check metrics with alerting
metrics = monitor_with_alerting()

# Get alert history
alert_history = alerting.get_alert_history()
print(f"\nTotal alerts: {len(alert_history)}")

spark.stop()

📈 Performance Metrics

Metric	Target	Warning	Critical	Action
Job Duration	< 10min	10-30min	> 30min	Optimize code, increase resources
Stage Failure Rate	< 1%	1-5%	> 5%	Check data skew, increase parallelism
Task Duration	< 5min	5-15min	> 15min	Repartition data, check skew
Heap Usage	< 70%	70-85%	> 85%	Increase memory, optimize code
GC Overhead	< 5%	5-10%	> 10%	Tune GC, increase memory
Shuffle Size	< 5GB	5-20GB	> 20GB	Optimize partitions, use broadcast
Cache Hit Ratio	> 95%	90-95%	< 90%	Tune caching strategy
Executor Utilization	> 80%	60-80%	< 60%	Increase parallelism, check skew

🏆 Best Practices

Enable event logging - Record all application events for historical analysis
Configure appropriate metrics sinks - Use console for development, external systems for production
Set up alerting - Define thresholds for critical metrics and implement notifications
Monitor regularly - Review performance metrics daily for production applications
Analyze trends - Track metrics over time to identify patterns and anomalies
Document baselines - Establish performance baselines for comparison
Use History Server - Analyze completed applications for optimization opportunities
Integrate with monitoring systems - Connect Spark metrics to enterprise monitoring platforms
Create dashboards - Build visual dashboards for easy performance monitoring
Review and optimize - Continuously improve monitoring based on feedback

🔗 Related Topics

17-cluster-management.mdx: Cluster resource monitoring
18-gc-tuning.mdx: Garbage collection monitoring and optimization
19-spark-submit.mdx: Deployment monitoring configuration
15-data-quality.mdx: Data quality monitoring and validation

20. Monitoring & Metrics in PySpark

20. Monitoring & Metrics in PySpark

DfSpark UI

DfMetrics System

Shuffle Read/Write Formula

ThStraggler Detection

🏗️ Monitoring Architecture

📊 Metrics Collection Flow

🔍 Performance Monitoring Dashboard

📚 Detailed Explanation

📊 Key Concepts Table

💻 Code Examples

Basic Monitoring Configuration

Custom Metrics Collection

Metrics Export to External System

Alerting System

📈 Performance Metrics

🏆 Best Practices

🔗 Related Topics

See Also

Need Expert PySpark Help?