← Back to Articles Observability

Centralized Pipeline Observability in Databricks

Oselio Candido · May 21, 2026 · 8 min read

Scattered driver logs and print() statements do not provide platform observability. When an unattended job fails hours after its cluster terminates, diagnosing the issue usually means hunting through individual task logs — assuming anything was printed at all.

To build true platform observability, execution telemetry must be queryable, persistent, and impossible to skip. We solved this by embedding an automated log writer directly into the platform's shared DataFrame write path (PipelineOrchestrator).

Why enforce telemetry in the write path?

Relying on developers to manually log execution details fails over time:

  • Documentation & conventions. Opt-in conventions suffer from drift as teams grow.
  • Databricks Jobs UI. Good for point-in-time debugging, but lacks business context (domain, layer, target table) and can't be queried across runs.
  • Per-domain audit tables. Moves fragmentation up one level, making cross-platform queries tedious.

Wrapping the write step. Every transformation notebook in the platform routes its output through a shared write helper. Instrumenting that single entry point ensures every write records telemetry automatically, requiring zero extra effort from pipeline authors.

Log table schema

Each entry in the centralized log table (PIPELINE_LOG_SCHEMA) records a single step execution: 

from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType,
    TimestampType, DateType, MapType,
)

PIPELINE_LOG_SCHEMA = StructType([
    # Business context
    StructField("domain",          StringType(),  True),
    StructField("corporate",       StringType(),  True),
    StructField("layer",           StringType(),  True),

    # Target location
    StructField("target_catalog",  StringType(),  True),
    StructField("target_schema",   StringType(),  True),
    StructField("target_table",    StringType(),  True),

    # Execution timing
    StructField("event_date",      DateType(),    True),
    StructField("start_timestamp", TimestampType(), True),
    StructField("end_timestamp",   TimestampType(), True),

    # Outcome & volume metrics
    StructField("status",          StringType(),  True),   # "SUCCESS" | "FAILED"
    StructField("error_message",   StringType(),  True),   # truncated to 2000 chars
    StructField("rows",            IntegerType(), True),   # Delta history metric

    # Identity & links
    StructField("job_run_id",      StringType(),  True),
    StructField("task_run_id",     StringType(),  True),
    StructField("url_run_links",   MapType(StringType(), StringType()), True),
    StructField("run_as",          StringType(),  True),

    # Runtime environment
    StructField("run_parameters",  MapType(StringType(), StringType()), True),
    StructField("commit_hash",     StringType(),  True),
    StructField("cluster_info",    MapType(StringType(), StringType()), True),
])

Implementation details

1. Fail-safe execution wrapper

The orchestrator executes the underlying Delta write inside a try/except/finally block: 

try:
    _run_write(config)
except Exception as exc:
    status = ExecutionStatus.FAILED
    exception = exc
    error_message = str(exc)[:2000]
else:
    status = ExecutionStatus.SUCCESS
    rows, _ = _get_delta_history_metrics(self._spark, full_table)
finally:
    end = datetime.now()
    event = PipelineLogEvent(...)
    self._writer.write_event(event)

if status == ExecutionStatus.FAILED:
    raise RuntimeError(f"Failed to process: {exception}") from exception
  • Guaranteed writes. Writing the telemetry event inside finally ensures failures leave an audit record even if the job crashes.
  • Exception re-raising. Re-raising the original exception preserves downstream job dependency chains rather than silently masking pipeline failures.

Slicing str(exc) to 2000 characters means a deep PySpark/JVM stack trace can get cut before it reaches the actual root cause buried at the bottom. That's an acceptable trade here rather than a gap, because error_message isn't the only way back to the failure — url_run_links is captured on the same row, so a truncated message is a pointer to go investigate, not the last available copy of the trace. The full log is still one click away in the Jobs UI.

2. Delta merge-aware row counts

Standard Delta write metrics report total output rows for append and overwrite operations, but omit row counts during MERGE INTO execution. Reading the latest commit from DESCRIBE HISTORY accounts for both insert and update operations: 

def _get_delta_history_metrics(spark, full_table_name: str) -> tuple[Optional[int], Dict[str, str]]:
    history = spark.sql(f"DESCRIBE HISTORY {full_table_name} LIMIT 1").first()
    metrics = history.operationMetrics or {}

    if "numOutputRows" in metrics:
        # append / overwrite
        rows = int(metrics["numOutputRows"])
    elif "numTargetRowsInserted" in metrics:
        # merge operations
        inserted = int(metrics.get("numTargetRowsInserted", 0))
        updated = int(metrics.get("numTargetRowsUpdated", 0))
        rows = inserted + updated
    else:
        rows = None

    return rows, metrics

3. Capturing context & run links

Execution parameters are harvested from Databricks widgets. To prevent log metadata extraction from breaking pipelines, widget parsing defaults gracefully when parameters are absent: 

def _get_widget_value(dbutils, widget_name: str, fallback: Optional[str] = None) -> Optional[str]:
    try:
        val = dbutils.widgets.get(widget_name)
        return val if val else fallback
    except Exception:
        return fallback

def _extract_org_id(host: Optional[str]) -> Optional[str]:
    if not host:
        return None
    match = re.search(r"adb-(\d+)", host)
    return match.group(1) if match else None

def _build_job_run_url(host: Optional[str], job_run_id: Optional[str], dbutils) -> Optional[str]:
    if not host or job_run_id is None:
        return None
    org_id = _extract_org_id(host)
    job_id = _get_job_id(dbutils)

    if not org_id or not job_id:
        return None

    return f"https://{host}/jobs/{job_id}/runs/{job_run_id}?o={org_id}"

4. Cluster metadata capture

Job performance often depends on hardware configuration. The log package queries the Databricks Clusters REST API (/api/2.0/clusters/get) to capture driver/executor node types, runtime versions, and cluster capacity: 

def fetch_cluster_info(dbutils, spark) -> Optional[ClusterInfo]:
    try:
        ctx = dbutils.notebook.entry_point.getDbutils().notebook().getContext()
        cluster_id = ctx.clusterId().get()
        token = ctx.apiToken().get()
    except Exception:
        return None

    host = spark.conf.get("spark.databricks.workspaceUrl", "") or None
    if not host or not cluster_id:
        return None

    try:
        resp = requests.get(
            f"https://{host}/api/2.0/clusters/get",
            headers={"Authorization": f"Bearer {token}"},
            params={"cluster_id": cluster_id},
            timeout=10,
        )
        resp.raise_for_status()
        return ClusterInfo.from_api_response(resp.json())
    except requests.RequestException:
        return None

5. Fail-open telemetry writes

Logging logic must never crash the primary data workload. If appending to the centralized log table fails, the exception is caught and logged to console output: 

def write_event(self, event: PipelineLogEvent) -> None:
    try:
        row = event.to_row()
        df = self._spark.createDataFrame([Row(**row)], schema=PIPELINE_LOG_SCHEMA)
        df.write.mode("append").saveAsTable(self._full_table_name)
    except Exception as e:
        print(f"[Telemetry] Failed to write log event: {e}")

Querying platform observability

Once every write records structured telemetry, SQL can answer cross-cutting operational questions.

Detecting duration drift

Identify tables with escalating runtime trends over time: 

SELECT
    target_table,
    event_date,
    AVG(unix_timestamp(end_timestamp) - unix_timestamp(start_timestamp)) AS avg_duration_seconds
FROM log_process
GROUP BY target_table, event_date
ORDER BY target_table, event_date;

Isolating volume outliers

Find successful runs where processed row counts exceed 3 standard deviations from historical averages: 

SELECT target_table, event_date, rows
FROM log_process
WHERE target_table = 'fact_orders'
  AND status = 'SUCCESS'
  AND rows > (
      SELECT AVG(rows) + 3 * STDDEV(rows)
      FROM log_process
      WHERE target_table = 'fact_orders' AND status = 'SUCCESS'
  );

Investigating failed tasks with direct links

Extract failure error messages alongside direct links to the relevant Databricks task execution UI: 

SELECT
    domain,
    layer,
    target_table,
    error_message,
    url_run_links['url_task_run_id'] AS failed_run_url
FROM log_process
WHERE status = 'FAILED'
  AND event_date >= date_sub(current_date(), 7)
ORDER BY start_timestamp DESC;

Trade-offs worth naming

None of these showed up as incidents on the infrastructure this ran on — a handful of notebooks, not a large fleet of concurrent tasks — but they're worth naming as the design's actual limits rather than pretending the approach has none.

Synchronous REST calls on the hot path

fetch_cluster_info calls https://{host}/api/2.0/clusters/get synchronously, via requests.get, on every tracked write. That's an outbound HTTP call to the Databricks control plane sitting on the critical path of a data write, and at a large enough scale — dozens of concurrent tasks hammering the same API — it's a plausible source of added latency or 429 Too Many Requests throttling. It didn't surface here because the write volume never got close to that; it would be the first thing to revisit before pointing this at a much larger job fleet, most likely by caching cluster info once per cluster lifetime instead of refetching it per write.

There's a second fragility in the same function: ctx.apiToken() reaches into an undocumented internal Java API. Hardened workspace configurations — Unity Catalog single-user clusters, restricted service principals — can return a null or blocked token for security reasons, which is exactly the kind of failure the try/except around it is built to absorb, but it's still a dependency on an API surface Databricks doesn't officially support.

What Databricks system tables already give you

Databricks' own system tables — system.lakeflow.jobs, system.compute.clusters, system.billing.usage — already record job execution, cluster specs, and usage platform-wide, with no custom code or API calls involved. Where those are enabled, a chunk of what cluster_info and the run-link construction do here duplicates something Databricks now ships natively; they weren't available in this workspace, which is why the package builds that context itself. Where system tables are available, the custom package's value narrows to what they can't give you: business-level context like target_table, rows, and domain, joined against your own write semantics rather than Databricks' generic job metadata.

Conclusion

Centralizing pipeline telemetry within the DataFrame write layer converts scattered log outputs into an operational dataset. By capturing target schemas, commit hashes, cluster specs, and execution metrics consistently across all runs, debugging shifts from reading unstructured driver logs to executing simple SQL queries.

Because it's one well-typed table instead of scattered print statements, it's also something dashboards and alerts can be built directly on top of — a chart of failure rate by domain and layer, or a scheduled query on status = 'FAILED' wired into a notification channel, without touching a single notebook.