The History Server is a standalone process that reads the event log directory and reconstructs the exact same Spark UI (Jobs, Stages, SQL, Storage, Environment tabs) for completed (and even still-running) applications. In Databricks/EMR/Synapse, this is built-in — you simply click on a finished job run to see its "Spark UI" tab, which is powered by the History Server reading event logs.

# spark-defaults.conf (on the History Server host)
spark.history.fs.logDirectory   s3a://my-bucket/spark-event-logs/
spark.history.ui.port           18080
spark.history.fs.update.interval  10s
spark.history.retainedApplications 200

# Start it
$SPARK_HOME/sbin/start-history-server.sh

# Browse to http://<host>:18080 to see all past applications

Because event logs are newline-delimited JSON, you can write custom scripts that scan thousands of historical jobs to build cost dashboards, detect regressions in stage durations, or find jobs with heavy spill — without opening the UI for each one.

import json, gzip

total_shuffle_read = 0
total_spill = 0
slow_tasks = []

with gzip.open("application_1700000000_0001.json.gz") as f:
    for line in f:
        event = json.loads(line)
        if event.get("Event") == "SparkListenerTaskEnd":
            metrics = event["Task Metrics"]
            total_shuffle_read += metrics["Shuffle Read Metrics"]["Remote Bytes Read"]
            total_spill += metrics.get("Disk Bytes Spilled", 0)
            duration = event["Task Info"]["Finish Time"] - event["Task Info"]["Launch Time"]
            if duration > 60_000:  # tasks over 1 minute
                slow_tasks.append((event["Task Info"]["Task ID"], duration))

print(f"Total shuffle read: {total_shuffle_read / 1e9:.2f} GB")
print(f"Total disk spill: {total_spill / 1e9:.2f} GB")
print(f"Slow tasks (>60s): {len(slow_tasks)}")

Each executor reports its own set of metrics: number of active/completed/failed tasks, bytes read/written, shuffle bytes, storage memory used vs available, execution memory used, and disk bytes used for spill. Since Spark 3.0, a richer set of "Executor Metrics" (peak JVM memory, peak on/off-heap storage memory, peak execution memory) is available via ExecutorMetricsSource.

Because Spark runs on the JVM, the JvmSource exposes standard JVM metrics for both driver and every executor: heap usage (used/committed/max for Eden, Survivor, Old Gen), non-heap usage (Metaspace, Code Cache), thread counts, and class loading stats. These come straight from Java's MemoryMXBean and ThreadMXBean.

GC metrics report the count and total time spent in young-generation and old-generation garbage collection per executor, sourced from GarbageCollectorMXBean. They are also surfaced directly in the Spark UI Tasks table as the "GC Time" column.

This is the OS-level memory used by the executor container — JVM heap + off-heap + Python process memory + OS overhead. On YARN/Kubernetes, if container memory exceeds the configured limit (spark.executor.memory + spark.executor.memoryOverhead), the container is killed by the resource manager, producing the dreaded "Container killed by YARN for exceeding memory limits" error — distinct from a JVM OutOfMemoryError.

Executors use local disk (configured via spark.local.dir) for shuffle files and spilled data when memory runs out. Monitoring disk usage and disk I/O per node reveals whether shuffle/spill is saturating local storage — a common cause of "No space left on device" failures during large joins or aggregations.

Shuffles, broadcast joins, and reads from remote object stores (S3, ADLS) all generate network traffic between nodes. Monitoring NetworkIn/NetworkOut per node highlights whether a stage is network-bound — common in large shuffles or when reading uncompressed data from S3 across availability zones.

Output Records / Output Size show how many rows and bytes each task wrote to the final sink (Parquet, Delta, JDBC). A sudden drop in output rows compared to input rows across a filter stage confirms your filter selectivity; an unexpectedly large number of small output files (one per task) is the classic "small files problem" caused by too many shuffle partitions relative to data size.

Shuffle Read breaks down into Local Bytes Read (data already on the same node — cheap) and Remote Bytes Read (fetched over the network — expensive). The "Shuffle Read Blocked Time" metric shows how long a task waited to fetch shuffle blocks from other executors — high values indicate network contention or too many small shuffle blocks.

Shuffle Write Records / Bytes are produced by the "map" side of a wide transformation — every task writes a shuffle file (partitioned by the shuffle key) for the next stage to read. A large Shuffle Write relative to Input Size means your transformation is "amplifying" data before the shuffle (e.g., an explode() before a groupBy) — consider filtering or pre-aggregating earlier.

When a task's in-memory data structure (sort buffer, hash table for aggregation/join) exceeds its allotted execution memory, Spark spills it to disk. "Spill (Memory)" is the size of the data before spilling (in deserialized form); "Spill (Disk)" is the size after being written to disk (serialized/compressed — usually smaller). Any non-zero spill is a signal of memory pressure for that task.

Task Duration is broken down by the UI into: Scheduler Delay (time waiting to be scheduled), Deserialization Time, Executor Compute Time (actual work), Shuffle Read/Write Time, and Result Serialization Time. Comparing the min/25th/median/75th/max columns in the stage summary is the fastest way to assess skew.

Prometheus needs to know where to scrape from. On Kubernetes, the standard approach is pod annotations that a Prometheus instance (often via the prometheus-operator) auto-discovers; on EMR/standalone clusters, you add static targets to prometheus.yml.

scrape_configs:
  - job_name: 'spark-driver'
    metrics_path: /metrics/prometheus
    static_configs:
      - targets: ['spark-driver-svc:4040']
        labels:
          app: 'pyspark-etl'
          env: 'prod'

  - job_name: 'spark-executors'
    metrics_path: /metrics/executors/prometheus
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_spark_role]
        regex: executor
        action: keep

Prometheus itself has a basic built-in graphing UI (useful for ad-hoc PromQL queries), but in production it almost always acts as the data source for Grafana, which provides richer dashboards, alerting, and templating. The architecture is: Spark exposes metrics → Prometheus scrapes & stores them → Grafana queries Prometheus and renders dashboards.

Grafana (and Prometheus Alertmanager) alert rules combine a PromQL query, a threshold, and an evaluation window. When the condition is true for the configured duration, an alert fires and is routed to a notification channel (Slack, PagerDuty, email).

groups:
  - name: spark-alerts
    rules:
      - alert: HighGCTime
        expr: rate(metrics_executor_jvm_G1_Old_Generation_time_Value[5m]) > 0.3
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Executor {{ $labels.instance }} spending >30% time in GC"

      - alert: ShufflePartitionSkew
        expr: max(metrics_stage_shuffleReadBytes_Value) / avg(metrics_stage_shuffleReadBytes_Value) > 10
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Possible data skew detected in shuffle stage"

Different metrics suit different visualizations: Time series for trends (heap usage, GC time over a job's lifetime), Gauges for current state (% of executors active vs requested), Bar gauges / heatmaps for task duration distributions across a stage, and Tables for per-job summaries (duration, records processed, status).

Because the same Datadog Agent also reports standard host metrics (CPU, memory, disk, network — the same categories from Module 26.4) tagged with the same cluster/host tags, Datadog lets you view Spark executor metrics side-by-side with the underlying EC2/Kubernetes node metrics in one dashboard — no manual joining of two separate monitoring systems.

Datadog "Monitors" are the equivalent of Prometheus alert rules — a metric query plus a threshold plus notification targets (Slack, PagerDuty, email, webhooks). Datadog also supports anomaly detection and forecast monitor types, which use historical patterns rather than fixed thresholds — useful for "job duration is unusually high for a Tuesday" style alerts.

{
  "name": "Spark Job Duration Anomaly",
  "type": "query alert",
  "query": "avg(last_4h):anomalies(avg:spark.job.duration{env:production}, 'basic', 2) >= 1",
  "message": "Spark job duration is anomalous for {{host.name}} @slack-data-eng-alerts",
  "options": { "thresholds": { "critical": 1 } }
}

Plain-text log lines like "Processed 1000 records in 5s" are hard to query at scale. Structured logging emits each log entry as JSON with consistent fields (timestamp, level, message, pipeline_name, run_id, row_count) — making them directly searchable/filterable in CloudWatch Logs Insights, Elasticsearch, or Datadog Logs without regex parsing.

import logging, json, sys, time

class JSONFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if hasattr(record, "extra_fields"):
            payload.update(record.extra_fields)
        return json.dumps(payload)

logger = logging.getLogger("pyspark_etl")
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

logger.info("Stage completed", extra={"extra_fields": {
    "pipeline_name": "orders_etl",
    "run_id": "r-20260615-001",
    "rows_written": 125_430,
    "duration_seconds": 42.7
}})
# Output: {"timestamp": "...", "level": "INFO", "logger": "pyspark_etl",
#          "message": "Stage completed", "pipeline_name": "orders_etl", ...}

A Correlation ID (often a run_id or batch_id) is generated once at the start of a pipeline run and attached to every log line, metric, and audit record produced during that run — across Airflow, Spark driver logs, executor logs, and any downstream system (Glue, Lambda). This lets you filter logs across multiple systems for one failed run using a single search term.

import argparse, uuid

parser = argparse.ArgumentParser()
parser.add_argument("--run-id", default=str(uuid.uuid4()))
args = parser.parse_args()

run_id = args.run_id
logger.info("Job started", extra={"extra_fields": {"run_id": run_id}})

# ... pipeline logic ...

logger.info("Job finished", extra={"extra_fields": {
    "run_id": run_id, "status": "SUCCESS", "rows_written": 125430
}})

Both Log4j and Python's logging module use the same severity hierarchy: DEBUG < INFO < WARN < ERROR < FATAL. Setting a logger to a given level shows that level and everything more severe. In production PySpark jobs, the default is almost always WARN for Spark internals (DAGScheduler, BlockManager are extremely chatty at INFO) and INFO for your own application logger.