Observability: структурированное логирование и трассировка Spark job

24/01/15 02:14:37 INFO DAGScheduler: Stage 12 (collect at OrdersJob.scala:87) finished in 14.8 s
24/01/15 02:14:38 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI
24/01/15 02:14:52 ERROR SparkContext: Error initializing SparkContext.
java.lang.OutOfMemoryError: GC overhead limit exceeded
    at org.apache.spark.sql.catalyst.expressions...
    (47 more frames)

2024-01-15 02:14:22 - Processing batch for date 2024-01-14, loaded 5234891 rows from Bronze
2024-01-15 02:15:44 - Deduplication complete: 5198234 unique rows remain
2024-01-15 02:17:01 - Writing to Silver...

{
  "timestamp": "2024-01-15T02:14:22.341Z",
  "level": "INFO",
  "pipeline": "bronze_to_silver_orders",
  "run_id": "airflow_run_2024-01-15T00:00:00",
  "stage": "bronze_read",
  "layer": "bronze",
  "table": "orders",
  "processing_date": "2024-01-14",
  "rows_read": 5234891,
  "duration_ms": 82341,
  "trace_id": "7f3a2b1c9d4e5f6a",
  "host": "spark-driver-xyz"
}

# log4j2.properties - добавить в spark.driver.extraJavaOptions
# или передать через --conf spark.driver.extraJavaOptions=-Dlog4j2.configurationFile=...

status = error
name = PropertiesConfig

# Appender: структурированный JSON в stdout
appender.console.type = Console
appender.console.name = JsonConsole
appender.console.layout.type = JsonTemplateLayout
appender.console.layout.eventTemplateUri = classpath:EcsLayout.json

# Корневой логгер
rootLogger.level = WARN
rootLogger.appenderRef.console.ref = JsonConsole

# Spark-специфичные логгеры
logger.spark.name = org.apache.spark
logger.spark.level = WARN

# Наш application-код логируем на INFO
logger.app.name = com.company.dataplatform
logger.app.level = INFO

spark-submit \
  --conf "spark.driver.extraJavaOptions=-Dlog4j2.configurationFile=/opt/spark/conf/log4j2.properties" \
  --conf "spark.executor.extraJavaOptions=-Dlog4j2.configurationFile=/opt/spark/conf/log4j2.properties" \
  jobs/silver_orders.py

pip install python-json-logger

import logging
import json
import os
import time
from datetime import datetime
from pythonjsonlogger import jsonlogger


class SparkPipelineLogger:
    """
    Структурированный логгер для PySpark-пайплайна.
    Каждая запись - валидный JSON с предустановленным контекстом.
    Контекст (pipeline_name, run_id, trace_id) устанавливается один раз
    при инициализации и автоматически добавляется ко всем записям.
    """

    def __init__(
        self,
        pipeline_name: str,
        run_id: str,
        trace_id: str = None,
        layer: str = None,
    ):
        self.pipeline_name = pipeline_name
        self.run_id = run_id
        self.trace_id = trace_id or self._generate_trace_id()
        self.layer = layer

        # Настраиваем logger с JSON форматтером
        self._logger = logging.getLogger(f"pipeline.{pipeline_name}")
        self._logger.setLevel(logging.DEBUG)

        if not self._logger.handlers:
            handler = logging.StreamHandler()
            formatter = jsonlogger.JsonFormatter(
                fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
                datefmt="%Y-%m-%dT%H:%M:%S.%fZ",
            )
            handler.setFormatter(formatter)
            self._logger.addHandler(handler)

    def _generate_trace_id(self) -> str:
        import uuid
        return uuid.uuid4().hex[:16]

    def _base_context(self) -> dict:
        """Контекст, который добавляется к каждой записи."""
        return {
            "pipeline": self.pipeline_name,
            "run_id": self.run_id,
            "trace_id": self.trace_id,
            "layer": self.layer,
            "host": os.environ.get("HOSTNAME", "unknown"),
            "spark_app_id": os.environ.get("SPARK_APP_ID", "unknown"),
        }

    def info(self, message: str, **extra):
        self._logger.info(message, extra={**self._base_context(), **extra})

    def warning(self, message: str, **extra):
        self._logger.warning(message, extra={**self._base_context(), **extra})

    def error(self, message: str, **extra):
        self._logger.error(message, extra={**self._base_context(), **extra})

    def stage_start(self, stage: str, **extra) -> float:
        """Логируем начало стадии и возвращаем timestamp для расчёта длительности."""
        self.info(
            f"Stage started: {stage}",
            event="stage_start",
            stage=stage,
            **extra,
        )
        return time.time()

    def stage_end(self, stage: str, start_time: float, **extra):
        """Логируем конец стадии с длительностью в миллисекундах."""
        duration_ms = int((time.time() - start_time) * 1000)
        self.info(
            f"Stage completed: {stage}",
            event="stage_end",
            stage=stage,
            duration_ms=duration_ms,
            **extra,
        )
        return duration_ms

    def rows_processed(
        self,
        stage: str,
        rows_in: int,
        rows_out: int,
        table: str = None,
        **extra,
    ):
        """Специализированный лог для статистики обработки строк."""
        self.info(
            f"Rows processed in {stage}",
            event="rows_processed",
            stage=stage,
            rows_in=rows_in,
            rows_out=rows_out,
            rows_dropped=rows_in - rows_out,
            drop_ratio=round((rows_in - rows_out) / rows_in, 4) if rows_in > 0 else 0,
            table=table,
            **extra,
        )

    def quality_gate(
        self,
        gate_name: str,
        passed: bool,
        metric_value,
        threshold,
        **extra,
    ):
        """Лог результата Quality Gate - для мониторинга и алертинга."""
        level = "info" if passed else "warning"
        getattr(self, level)(
            f"Quality gate {'PASSED' if passed else 'FAILED'}: {gate_name}",
            event="quality_gate",
            gate_name=gate_name,
            passed=passed,
            metric_value=metric_value,
            threshold=threshold,
            **extra,
        )

def run_bronze_to_silver(
    spark: SparkSession,
    batch_date: str,
    run_id: str,
    trace_id: str = None,
) -> None:
    log = SparkPipelineLogger(
        pipeline_name="bronze_to_silver_orders",
        run_id=run_id,
        trace_id=trace_id,
        layer="silver",
    )

    # ── Шаг 1: чтение Bronze ──
    t0 = log.stage_start("bronze_read", table="orders", date=batch_date)

    bronze_df = spark.read.format("delta").load(
        f"s3a://datalake/bronze/orders/date={batch_date}/"
    )
    bronze_count = bronze_df.count()

    log.stage_end("bronze_read", t0, rows_read=bronze_count, table="orders")
    log.rows_processed("bronze_read", rows_in=bronze_count, rows_out=bronze_count)

    # ── Шаг 2: дедупликация ──
    t1 = log.stage_start("deduplication")
    deduped_df = bronze_df.dropDuplicates(["order_id"])
    deduped_count = deduped_df.count()
    dup_count = bronze_count - deduped_count

    log.stage_end("deduplication", t1, rows_deduplicated=dup_count)
    log.rows_processed(
        "deduplication",
        rows_in=bronze_count,
        rows_out=deduped_count,
        table="orders",
    )

    # ── Шаг 3: quality gate ──
    null_count = deduped_df.filter(F.col("order_id").isNull()).count()
    log.quality_gate(
        gate_name="null_order_id_check",
        passed=(null_count == 0),
        metric_value=null_count,
        threshold=0,
        table="orders",
    )

    if null_count > 0:
        raise ValueError(f"Null order_id detected: {null_count} rows")

    # ── Шаг 4: запись Silver ──
    t2 = log.stage_start("silver_write", table="orders")
    (
        deduped_df
        .write.mode("overwrite")
        .format("delta")
        .partitionBy("processing_date")
        .save("s3a://datalake/silver/orders/")
    )
    log.stage_end("silver_write", t2, rows_written=deduped_count)

    log.info(
        "Pipeline complete",
        event="pipeline_complete",
        bronze_rows=bronze_count,
        silver_rows=deduped_count,
        duplicates_removed=dup_count,
    )

{
  "timestamp": "2024-01-15T02:15:44.891Z",
  "level": "INFO",
  "event": "rows_processed",
  "message": "Rows processed in deduplication",

  "pipeline": "bronze_to_silver_orders",
  "run_id": "scheduled__2024-01-15T00:00:00+00:00",
  "trace_id": "7f3a2b1c9d4e5f6a",
  "layer": "silver",
  "stage": "deduplication",
  "table": "orders",

  "rows_in": 5234891,
  "rows_out": 5198234,
  "rows_dropped": 36657,
  "drop_ratio": 0.007,

  "host": "spark-driver-orders-abc123",
  "spark_app_id": "application_1705276800000_0042"
}

Airflow DAG: daily_orders → запускает task: extract_orders
extract_orders: spark-submit jobs/extract.py → Application ID: app_001
             → Stage 0: read Kafka (OK)
             → Stage 1: write Bronze (OK)

Airflow DAG: daily_orders → запускает task: transform_silver
transform_silver: spark-submit jobs/transform.py → Application ID: app_002
               → Stage 0: read Bronze (OK)
               → Stage 1: dedup (FAILED after 14 min)

# dags/orders_pipeline.py

import uuid
from airflow.models import Variable
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

def generate_trace_id(**context) -> str:
    """
    Генерируем trace_id один раз для всего DAG-запуска.
    Используем комбинацию dag_run_id + короткий UUID для читаемости.
    """
    dag_run_id = context["run_id"]
    short_uuid = uuid.uuid4().hex[:8]
    # Формат: airflow_run_id + короткий суффикс
    return f"{dag_run_id[:20]}_{short_uuid}"


with DAG(dag_id="orders_pipeline", ...) as dag:

    # Генерация trace_id через BashOperator или XCom
    trace_id = "{{ run_id | replace(':', '') | replace('+', '') | truncate(20, False, '') }}_{{ macros.uuid.uuid4().hex[:8] }}"

    extract_task = SparkSubmitOperator(
        task_id="extract_orders",
        application="jobs/extract_orders.py",
        application_args=["--date", "{{ ds }}", "--run-id", "{{ run_id }}"],
        conf={
            # Trace ID передаётся как Spark conf - доступен в коде через SparkConf
            "spark.app.trace_id": "{{ run_id }}",
            "spark.app.pipeline_name": "orders_pipeline",
            "spark.app.dag_run_id": "{{ run_id }}",
            "spark.app.task_id": "extract_orders",
        },
    )

    transform_task = SparkSubmitOperator(
        task_id="transform_silver",
        application="jobs/transform_silver.py",
        application_args=["--date", "{{ ds }}", "--run-id", "{{ run_id }}"],
        conf={
            "spark.app.trace_id": "{{ run_id }}",
            "spark.app.pipeline_name": "orders_pipeline",
            "spark.app.task_id": "transform_silver",
        },
    )

    extract_task >> transform_task

from pyspark.sql import SparkSession
from pyspark import SparkContext


def setup_trace_context(spark: SparkSession) -> dict:
    """
    Читаем trace_id и другой контекст из SparkConf
    (куда Airflow передал их через --conf).
    Затем устанавливаем в LocalProperties для автоматической
    передачи во все Executor Task'и.
    """
    sc: SparkContext = spark.sparkContext
    conf = sc.getConf()

    trace_id = conf.get("spark.app.trace_id", "no_trace")
    pipeline_name = conf.get("spark.app.pipeline_name", "unknown")
    dag_run_id = conf.get("spark.app.dag_run_id", "unknown")
    task_id = conf.get("spark.app.task_id", "unknown")

    # Устанавливаем в LocalProperties Spark —
    # это магия: свойства автоматически копируются в каждую Task на Executor'е
    sc.setLocalProperty("trace_id", trace_id)
    sc.setLocalProperty("pipeline_name", pipeline_name)
    sc.setLocalProperty("dag_run_id", dag_run_id)
    sc.setLocalProperty("task_id", task_id)

    # Spark UI тоже умеет показывать эти свойства в Job Description
    sc.setJobDescription(
        f"pipeline={pipeline_name} | run={dag_run_id[:20]} | trace={trace_id[:8]}"
    )

    return {
        "trace_id": trace_id,
        "pipeline_name": pipeline_name,
        "dag_run_id": dag_run_id,
        "task_id": task_id,
        "spark_app_id": sc.applicationId,
    }


def main():
    spark = SparkSession.builder.appName("TransformSilverOrders").getOrCreate()

    # Устанавливаем контекст трассировки
    ctx = setup_trace_context(spark)

    # Создаём логгер с trace_id
    log = SparkPipelineLogger(
        pipeline_name=ctx["pipeline_name"],
        run_id=ctx["dag_run_id"],
        trace_id=ctx["trace_id"],
        layer="silver",
    )

    log.info(
        "Job started",
        event="job_start",
        spark_app_id=ctx["spark_app_id"],
        task_id=ctx["task_id"],
    )

    # ... дальнейшая логика пайплайна

from pyspark import TaskContext
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType


def create_traced_udf(func):
    """
    Обёртка для UDF, которая добавляет trace_id из TaskContext в логи.
    TaskContext.get().getLocalProperty("trace_id") возвращает значение,
    которое было установлено на Driver через sc.setLocalProperty().
    """
    import logging

    def traced_func(*args):
        # Получаем trace_id из контекста текущей Task
        ctx = TaskContext.get()
        trace_id = ctx.getLocalProperty("trace_id") if ctx else "no_trace"
        partition_id = ctx.partitionId() if ctx else -1

        try:
            return func(*args)
        except Exception as e:
            # Логируем с trace_id - теперь можно найти по ID в ELK
            logger = logging.getLogger("executor.udf")
            logger.error(
                f"UDF failed | trace_id={trace_id} | "
                f"partition={partition_id} | error={str(e)}"
            )
            return None

    return traced_func


@create_traced_udf
def normalize_phone(phone: str) -> str:
    """Нормализация номера телефона с трассировкой."""
    import re
    if not phone:
        return None
    digits = re.sub(r"[^\d]", "", phone)
    return f"+7{digits[-10:]}" if len(digits) >= 10 else None


normalize_phone_udf = udf(normalize_phone, StringType())

from pyspark.sql import SparkSession
from pyspark.sql.execution.listener import QueryExecutionListener
import json
import logging

logger = logging.getLogger("spark.query_listener")


class DataPipelineQueryListener(QueryExecutionListener):
    """
    Кастомный QueryExecutionListener для сбора метрик каждой Spark SQL операции.

    Автоматически срабатывает после каждого .write(), .count() и других Actions.
    Не требует изменения кода пайплайна - подключается один раз при инициализации.
    """

    def __init__(self, pipeline_name: str, run_id: str, trace_id: str = None):
        self.pipeline_name = pipeline_name
        self.run_id = run_id
        self.trace_id = trace_id or "no_trace"

    def onSuccess(self, funcName: str, qe, durationNs: int) -> None:
        """
        Вызывается после успешного завершения любого Action.
        qe - QueryExecution объект с доталями физического плана и метриками.
        durationNs - длительность выполнения в наносекундах.
        """
        duration_ms = durationNs // 1_000_000

        # Извлекаем метрики из физического плана
        metrics = self._extract_write_metrics(qe)

        log_record = {
            "event": "query_execution_success",
            "pipeline": self.pipeline_name,
            "run_id": self.run_id,
            "trace_id": self.trace_id,
            "func_name": funcName,
            "duration_ms": duration_ms,
            **metrics,
        }

        logger.info(json.dumps(log_record))

    def onFailure(self, funcName: str, qe, exception: Exception) -> None:
        """
        Вызывается при ошибке любого Action.
        Логируем детали ошибки с полным контекстом.
        """
        log_record = {
            "event": "query_execution_failure",
            "pipeline": self.pipeline_name,
            "run_id": self.run_id,
            "trace_id": self.trace_id,
            "func_name": funcName,
            "error_type": type(exception).__name__,
            "error_message": str(exception)[:500],
        }
        logger.error(json.dumps(log_record))

    def _extract_write_metrics(self, qe) -> dict:
        """
        Извлекаем метрики из выполненного плана.
        Ищем WriteFilesExec или InsertIntoHadoopFsRelation для операций записи.
        """
        metrics = {}
        try:
            # Получаем все метрики из Spark Plan
            executed_plan = qe.executedPlan()
            for node in executed_plan:
                node_name = node.__class__.__name__

                # Метрики для операций записи в файлы
                if "WriteFiles" in node_name or "InsertInto" in node_name:
                    spark_metrics = node.metrics()
                    if spark_metrics:
                        metrics["num_output_rows"] = (
                            spark_metrics.get("numOutputRows", {}).get("value", 0)
                        )
                        metrics["num_output_bytes"] = (
                            spark_metrics.get("numOutputBytes", {}).get("value", 0)
                        )
                        metrics["num_parts"] = (
                            spark_metrics.get("numParts", {}).get("value", 0)
                        )
        except Exception:
            pass  # Метрики опциональны - не ломаем пайплайн при их отсутствии

        return metrics


def register_pipeline_listener(
    spark: SparkSession,
    pipeline_name: str,
    run_id: str,
    trace_id: str = None,
) -> DataPipelineQueryListener:
    """
    Регистрируем listener в SparkSession.
    После этого все Actions автоматически логируются.
    """
    listener = DataPipelineQueryListener(pipeline_name, run_id, trace_id)
    spark.streams.addListener(listener)  # для Structured Streaming
    spark._jvm.org.apache.spark.sql.execution.QueryExecution.listeners().add(
        listener._jlistener  # для Batch DataFrame
    )
    return listener

def register_task_metrics_listener(spark: SparkSession, log: SparkPipelineLogger):
    """
    Регистрируем SparkListener через Py4J для мониторинга Task-метрик.
    Позволяет детектировать data skew и GC pressure без Spark UI.
    """
    from py4j.java_gateway import java_import

    sc = spark.sparkContext
    jvm = sc._jvm
    jsc = sc._jsc

    java_import(jvm, "org.apache.spark.scheduler.*")

    class TaskMetricsListener:
        """Python-обёртка над Java SparkListener."""

        class Java:
            implements = ["org.apache.spark.scheduler.SparkListener"]

        def onTaskEnd(self, task_end_event):
            task_info = task_end_event.taskInfo()
            task_metrics = task_end_event.taskMetrics()

            if task_metrics is None:
                return

            # Ключевые метрики каждой Task
            shuffle_read = task_metrics.shuffleReadMetrics().totalBytesRead()
            shuffle_write = task_metrics.shuffleWriteMetrics().bytesWritten()
            spill_disk = task_metrics.diskBytesSpilled()
            spill_memory = task_metrics.memoryBytesSpilled()
            gc_time = task_metrics.jvmGCTime()
            duration = task_info.duration()

            # Детектируем аномалии
            if spill_disk > 1024 * 1024 * 1024:  # spill > 1GB
                log.warning(
                    "Large disk spill detected",
                    event="disk_spill_alert",
                    task_id=task_info.taskId(),
                    stage_id=task_end_event.stageId(),
                    spill_disk_bytes=spill_disk,
                    spill_gb=round(spill_disk / (1024**3), 2),
                )

            if gc_time > 30_000:  # GC > 30 секунд
                log.warning(
                    "Excessive GC time",
                    event="gc_pressure_alert",
                    task_id=task_info.taskId(),
                    gc_time_ms=gc_time,
                    task_duration_ms=duration,
                    gc_ratio=round(gc_time / duration, 2) if duration > 0 else 0,
                )

    listener = TaskMetricsListener()
    jsc.sc().addSparkListener(listener)
    return listener

def detect_data_skew(
    spark: SparkSession,
    app_id: str,
    stage_id: int,
    skew_ratio_threshold: float = 5.0,
) -> dict:
    """
    Анализируем Task-метрики конкретного Stage для детекции data skew.
    Skew определяем как: max_task_duration / avg_task_duration > threshold.

    В production используем Spark History Server REST API или SparkListener.
    Здесь показана логика обнаружения.
    """
    # В реальной системе Task-метрики получаем из Spark History Server API:
    # GET /api/v1/applications/{appId}/stages/{stageId}/taskList

    # Для демонстрации - псевдокод с реальными именами метрик
    task_durations = []  # заполняется из Task End Events

    if not task_durations:
        return {"status": "no_data"}

    max_duration = max(task_durations)
    avg_duration = sum(task_durations) / len(task_durations)
    min_duration = min(task_durations)
    skew_ratio = max_duration / avg_duration if avg_duration > 0 else 0

    result = {
        "stage_id": stage_id,
        "task_count": len(task_durations),
        "max_duration_ms": max_duration,
        "avg_duration_ms": round(avg_duration),
        "min_duration_ms": min_duration,
        "skew_ratio": round(skew_ratio, 2),
        "skew_detected": skew_ratio > skew_ratio_threshold,
    }

    if result["skew_detected"]:
        print(
            f"[SKEW ALERT] Stage {stage_id}: max={max_duration}ms, "
            f"avg={avg_duration:.0f}ms, ratio={skew_ratio:.1f}x. "
            f"Consider salting or repartitioning."
        )

    return result

# spark-defaults.conf или --conf при spark-submit

# Включаем Prometheus endpoint на Driver
spark.ui.prometheus.enabled=true

# Spark metrics system
spark.metrics.conf.*.sink.prometheus.class=org.apache.spark.metrics.sink.PrometheusServlet
spark.metrics.conf.*.sink.prometheus.path=/metrics/prometheus

# Метрики JVM (GC, heap, threads)
spark.metrics.conf.driver.source.jvm.class=org.apache.spark.metrics.source.JvmSource
spark.metrics.conf.executor.source.jvm.class=org.apache.spark.metrics.source.JvmSource

# HELP spark_driver_jvm_heap_used_bytes JVM heap used
# TYPE spark_driver_jvm_heap_used_bytes gauge
spark_driver_jvm_heap_used_bytes{app_id="application_001"} 2147483648

# HELP spark_executor_shuffleRead_totalBytesRead Total shuffle read bytes
# TYPE spark_executor_shuffleRead_totalBytesRead counter
spark_executor_shuffleRead_totalBytesRead{executor_id="1"} 53687091200

from pyspark import SparkContext


class SparkCustomMetrics:
    """
    Регистрируем кастомные метрики в Spark Metrics Registry.
    Они автоматически экспортируются через PrometheusServlet.
    """

    def __init__(self, sc: SparkContext, namespace: str):
        self.sc = sc
        self.namespace = namespace
        # Получаем Java Metrics Registry через Py4J
        self._registry = sc._jvm.org.apache.spark.metrics.MetricsSystem

    def register_gauge(self, name: str, value_func):
        """Регистрируем Gauge - метрику, читаемую в момент опроса."""
        # Через Py4J создаём Java Gauge
        jvm = self.sc._jvm
        gauge = jvm.com.codahale.metrics.Gauge.of(value_func)
        # self._registry.registerGauge(f"{self.namespace}.{name}", gauge)

    def increment_counter(self, name: str, amount: int = 1):
        """Инкремент счётчика - монотонно возрастающая метрика."""
        pass  # В реальной реализации через Py4J


def report_pipeline_metrics(
    spark: SparkSession,
    pipeline_name: str,
    stats: dict,
) -> None:
    """
    Отправляем кастомные метрики через statsd или напрямую в Prometheus
    (через pushgateway для batch jobs).
    """
    import socket
    import time

    # Формат statsd: metric_name:value|type
    metrics_lines = [
        f"spark.pipeline.rows_processed:{stats['total_rows']}|g",
        f"spark.pipeline.rows_clean:{stats['clean_rows']}|g",
        f"spark.pipeline.quarantine_rate:{stats['quarantine_rate']:.4f}|g",
        f"spark.pipeline.duration_seconds:{stats['duration_seconds']:.1f}|g",
    ]

    # Тэги для идентификации метрик
    tags = f",pipeline={pipeline_name},env=prod"

    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        for line in metrics_lines:
            message = f"{line}{tags}".encode()
            sock.sendto(message, ("statsd-host", 8125))
        sock.close()
    except Exception as e:
        print(f"[Metrics] Failed to send to statsd: {e}")

# Время выполнения последних 10 запусков пайплайна
avg_over_time(
  spark_pipeline_duration_seconds{pipeline="bronze_to_silver_orders"}[24h]
)

# Quarantine rate - тренд за 7 дней (алерт при росте)
spark_pipeline_quarantine_rate{pipeline="bronze_to_silver_orders"}
  > 0.05  # alert threshold

# Shuffle bytes как индикатор data skew
sum by (stage_id) (
  spark_executor_shuffleRead_totalBytesRead
)

# GC pressure на Executor'ах (алерт > 20% времени в GC)
sum(rate(spark_executor_jvm_gc_time_total[5m])) /
sum(rate(spark_executor_duration_total[5m])) > 0.20

# Количество spill to disk (должно быть 0 в нормальной ситуации)
sum(spark_executor_diskBytesSpilled_total)

from dataclasses import dataclass, field
from typing import Optional


@dataclass
class LayerObservabilityEvent:
    """Структурированное событие observability для конкретного слоя."""
    pipeline: str
    run_id: str
    trace_id: str
    layer: str          # bronze | silver | gold
    table: str
    processing_date: str
    timestamp: str

    # Bronze-специфичные метрики
    rows_ingested: Optional[int] = None
    bytes_ingested: Optional[int] = None
    parse_errors: Optional[int] = None
    schema_drift_detected: Optional[bool] = None
    source_lag_seconds: Optional[int] = None  # задержка от источника

    # Silver-специфичные метрики
    rows_before_dedup: Optional[int] = None
    rows_after_dedup: Optional[int] = None
    duplicates_removed: Optional[int] = None
    quarantine_rows: Optional[int] = None
    quarantine_rate: Optional[float] = None
    null_violations: Optional[dict] = None

    # Gold-специфичные метрики
    aggregation_groups: Optional[int] = None
    metric_delta_pct: Optional[float] = None  # % изменения ключевой метрики
    freshness_lag_minutes: Optional[int] = None  # задержка данных

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from datetime import datetime, timedelta


def check_data_freshness(
    spark: SparkSession,
    table_path: str,
    event_time_col: str,
    max_lag_minutes: int = 30,
    log: SparkPipelineLogger = None,
) -> dict:
    """
    Проверяем свежесть данных в таблице.
    Сравниваем MAX(event_time) с текущим временем.
    Если разница > max_lag_minutes - алерт.
    """
    result = (
        spark.read.format("delta").load(table_path)
        .agg(
            F.max(event_time_col).alias("latest_event"),
            F.min(event_time_col).alias("earliest_event"),
            F.count("*").alias("total_rows"),
        )
        .collect()[0]
    )

    latest_event = result["latest_event"]
    now = datetime.utcnow()

    lag_minutes = None
    if latest_event:
        lag_seconds = (now - latest_event.replace(tzinfo=None)).total_seconds()
        lag_minutes = int(lag_seconds / 60)

    freshness_ok = lag_minutes is not None and lag_minutes <= max_lag_minutes

    if log:
        log.info(
            "Data freshness check",
            event="freshness_check",
            table=table_path,
            latest_event=str(latest_event),
            lag_minutes=lag_minutes,
            threshold_minutes=max_lag_minutes,
            freshness_ok=freshness_ok,
        )

        if not freshness_ok:
            log.warning(
                "Data freshness SLA violated",
                event="freshness_sla_violation",
                table=table_path,
                lag_minutes=lag_minutes,
                threshold_minutes=max_lag_minutes,
            )

    return {
        "latest_event": str(latest_event),
        "lag_minutes": lag_minutes,
        "freshness_ok": freshness_ok,
        "total_rows": result["total_rows"],
    }

{pipeline="bronze_to_silver_orders"} |= "2024-01-15"

{"event": "stage_end", "stage": "bronze_read", "rows_read": 52348910, "duration_ms": 82341}
{"event": "rows_processed", "stage": "deduplication", "rows_in": 52348910, ...}
{"event": "query_execution_failure", "error_type": "OutOfMemoryError", "func_name": "save"}

Filter (isnotnull(processing_date#12))
  Scan parquet datalake/bronze/orders
    PartitionFilters: []
    PushedFilters: [IsNotNull(processing_date)]

def run_observable_pipeline(
    spark: SparkSession,
    batch_date: str,
    run_id: str,
) -> None:
    """
    Демонстрация полного observability-стека в одном пайплайне:
    - Structured logging с JSON
    - Correlation ID / trace propagation
    - SparkContext local properties
    - Автоматические метрики через QueryExecutionListener
    - Data freshness check
    - Business metrics logging
    """
    import time
    from datetime import datetime

    # 1. Инициализация trace context
    ctx = setup_trace_context(spark)

    log = SparkPipelineLogger(
        pipeline_name="observable_orders_pipeline",
        run_id=run_id,
        trace_id=ctx["trace_id"],
        layer="silver",
    )

    # 2. Регистрируем QueryExecutionListener
    register_pipeline_listener(
        spark,
        pipeline_name="observable_orders_pipeline",
        run_id=run_id,
        trace_id=ctx["trace_id"],
    )

    pipeline_start = time.time()
    log.info(
        "Observable pipeline started",
        event="pipeline_start",
        spark_app_id=ctx["spark_app_id"],
        batch_date=batch_date,
    )

    # 3. Bronze read с логированием
    t0 = log.stage_start("bronze_read", batch_date=batch_date)
    bronze_df = spark.read.format("delta").load(
        f"s3a://datalake/bronze/orders/date={batch_date}/"
    )
    bronze_count = bronze_df.count()
    log.stage_end("bronze_read", t0, rows_read=bronze_count)

    # 4. Quality check с логированием
    log.quality_gate(
        gate_name="bronze_row_count",
        passed=(bronze_count > 0),
        metric_value=bronze_count,
        threshold=1,
    )

    # 5. Transform с логированием каждого шага
    t1 = log.stage_start("deduplication")
    silver_df = bronze_df.dropDuplicates(["order_id"])
    silver_count = silver_df.count()
    log.stage_end("deduplication", t1, rows_deduplicated=bronze_count - silver_count)
    log.rows_processed("deduplication", bronze_count, silver_count, table="orders")

    # 6. Write - QueryExecutionListener логирует автоматически
    t2 = log.stage_start("silver_write")
    (
        silver_df
        .write.mode("overwrite")
        .format("delta")
        .partitionBy("processing_date")
        .save("s3a://datalake/silver/orders/")
    )
    log.stage_end("silver_write", t2, rows_written=silver_count)

    # 7. Freshness check
    freshness = check_data_freshness(
        spark,
        "s3a://datalake/silver/orders/",
        event_time_col="created_at",
        max_lag_minutes=120,
        log=log,
    )

    # 8. Итоговая метрика пайплайна
    pipeline_duration = time.time() - pipeline_start
    log.info(
        "Pipeline complete",
        event="pipeline_complete",
        duration_seconds=round(pipeline_duration, 1),
        bronze_rows=bronze_count,
        silver_rows=silver_count,
        dedup_ratio=round(1 - silver_count / bronze_count, 4) if bronze_count else 0,
        data_freshness_ok=freshness["freshness_ok"],
        freshness_lag_minutes=freshness["lag_minutes"],
    )

    # 9. Отправка business-метрик в Prometheus/StatsD
    report_pipeline_metrics(
        spark,
        pipeline_name="observable_orders_pipeline",
        stats={
            "total_rows": bronze_count,
            "clean_rows": silver_count,
            "quarantine_rate": 0.0,
            "duration_seconds": pipeline_duration,
        },
    )

# АНТИПАТТЕРН: логируем каждую строку в цикле
for row in df.collect():  # уже антипаттерн - collect на большом DF
    logger.debug(f"Processing row: {row}")  # миллионы строк в логах

# ПРАВИЛЬНО: агрегированные метрики, не отдельные строки
logger.info("Batch processed", rows=df.count(), stage="transform")

# АНТИПАТТЕРН: каждый шаг логирует независимо
# В ELK невозможно связать logs одного pipeline run
logger.info("Bronze read complete: 5M rows")  # нет run_id, нет trace_id
logger.info("Silver write complete: 4.9M rows")  # это тот же запуск? другой?

# ПРАВИЛЬНО: единый trace_id через всю цепочку
log.info("Bronze read complete", trace_id=trace_id, rows=5_000_000)
log.info("Silver write complete", trace_id=trace_id, rows=4_900_000)

# АНТИПАТТЕРН: алерт на каждое небольшое отклонение
if current_rows < previous_rows:  # срабатывает при любом снижении, даже на 1 строку
    send_pagerduty_alert("Row count decreased!")

# ПРАВИЛЬНО: осмысленный порог с контекстом
drop_ratio = (previous_rows - current_rows) / previous_rows
if drop_ratio > 0.30:  # > 30% снижение - реальная проблема
    send_alert(
        level="critical",
        message=f"Row count dropped by {drop_ratio:.1%}",
        context={"previous": previous_rows, "current": current_rows},
    )

# ПРАВИЛЬНЫЙ подход: минимальный, но полезный observability stack

ESSENTIAL_SIGNALS = {
    # Один структурированный лог с trace_id - лучше тысячи разрозненных
    "logging": "structured JSON with trace_id",

    # Три ключевые метрики на пайплайн - лучше 50 технических
    "metrics": [
        "pipeline_duration_seconds",    # как долго?
        "rows_quarantine_rate",         # насколько чисты данные?
        "data_freshness_lag_minutes",   # насколько свежи данные?
    ],

    # Один алерт на pipeline - лучше 20 шумных
    "alerting": "quarantine_rate > 5% OR duration > 2x_sla OR freshness_lag > 2h",
}

# Признак правильной архитектуры: каждая функция возвращает observable результат

# ПЛОХО: функция ничего не возвращает, observability невозможна
def process_bronze(df):
    df.write.parquet(silver_path)
    # void - не знаем, сколько строк записано

# ХОРОШО: функция возвращает статистику, observability встроена
def process_bronze(df, log: SparkPipelineLogger) -> dict:
    t = log.stage_start("write_silver")
    (df.write.format("delta").save(silver_path))
    count = df.count()
    log.stage_end("write_silver", t, rows_written=count)
    return {"rows_written": count, "path": silver_path}

OBSERVABILITY_CHECKLIST = {
    "structured_logging": [
        "JSON-формат для всех логов (не plain text)",
        "Каждое событие содержит: pipeline, run_id, trace_id, timestamp",
        "Числовые метрики в полях (rows_read, duration_ms) - не в строках",
        "log.stage_start()/stage_end() на каждую значимую операцию",
    ],
    "tracing": [
        "trace_id передаётся из Airflow через --conf",
        "sc.setLocalProperty('trace_id', ...) вызывается при старте",
        "Python UDF имеют доступ к trace_id через TaskContext",
        "Все Spark jobs описаны: sc.setJobDescription()",
    ],
    "metrics": [
        "Итоговые метрики пайплайна логируются в pipeline_complete event",
        "Quarantine rate присутствует если есть DLQ",
        "Data freshness проверяется на Gold слое",
        "Метрики отправляются в Prometheus/StatsD",
    ],
    "alerting": [
        "Алерт при quarantine_rate > 5%",
        "Алерт при duration > 2x исторического среднего",
        "Алерт при data freshness lag > SLA",
        "Алерт при пустом батче (rows == 0)",
        "Алерт не чаще 1 раза в 15 минут (группировка)",
    ],
    "debugging": [
        "Spark History Server настроен и хранит event logs > 30 дней",
        "По trace_id можно найти все логи запуска в ELK/Loki",
        "По application_id можно найти Stage/Task метрики в History Server",
        "QueryExecutionListener зарегистрирован для автоматических write-метрик",
    ],
}