Data Quality, Backlog and Fleet Monitoring with Pipeline Events

1.Expectation metrics in flow_progress events

The queries on this page run against event_log_raw, a view you create over a pipeline's event log, either the published table or event_log(<pipelineId>). If your datasets have expectations, the event log records how many rows met each one.

The metrics are stored at two levels. Per-expectation counts are in details:flow_progress.data_quality.expectations, an array with one entry per expectation that holds its name, dataset, passed_records and failed_records. The total of dropped rows for the flow is one level up, in details:flow_progress.data_quality.dropped_records.

Because the expectations field is a JSON array, the docs' query parses it with from_json and then explodes it into one row per expectation. It filters on event_type = 'flow_progress' and the latest update_id, then groups by dataset and expectation name to sum the passing and failing records:

explode(
        from_json(
          details:flow_progress:data_quality:expectations,
          "array<struct<name: string, dataset: string, passed_records: int, failed_records: int>>"
        )
      ) row_expectations

The action on an expectation decides what happens to invalid rows. Retaining invalid records is the default (warn) behavior, written with the expect operator: records that violate the expectation are added to the target dataset along with valid records, and metrics on how many records pass or fail are still collected. The expect_or_drop operator drops violating records from the target dataset instead. The fail action stops the update, so no metrics are recorded for it.

There are two gaps to know about. First, an expectation with the fail action stops the update as soon as an invalid record arrives, so no metrics are recorded for it. Second, metrics may be missing for some datasets. Streaming tables and materialized views backed by a standalone pipeline created in Databricks SQL have no Data quality tab in the pipeline UI, so the docs tell you to query the event log to view their expectation metrics.

Sources12

2.Throughput, backlog and Auto Loader progress

flow_progress events also measure how much work each flow does. details:flow_progress.metrics holds num_output_rows, num_upserted_rows and num_deleted_rows, and details:flow_progress.status follows the flow from STARTING and RUNNING to a final status such as COMPLETED, FAILED, SKIPPED or IDLE. The docs' all-flows query aggregates these per update_id and table to report start time, end time, duration, final status and row totals for each flow. Streaming lag is reported in details:flow_progress.metrics.backlog_bytes:

with latest_update as (
  SELECT origin.update_id as id
    FROM event_log_raw
    WHERE event_type = 'create_update'
    ORDER BY timestamp DESC
    limit 1 -- remove if you want all of the update_ids
)
SELECT
  timestamp,
  Double(details :flow_progress.metrics.backlog_bytes) as backlog
FROM
  event_log_raw,
  latest_update
WHERE
  event_type ='flow_progress'
  AND
  origin.update_id = latest_update.id;

Auto Loader file processing creates operation_progress events whose details:operation_progress.type is AUTO_LOADER_LISTING or AUTO_LOADER_BACKFILL. These include status, duration_ms, the source path and the number of files listed. On classic compute with enhanced autoscaling, cluster resizes are logged as autoscale events. For lineage, use flow_definition events: their details:flow_definition object lists the output_dataset and input_datasets of each flow.

Sources31

3.Monitoring every pipeline with the pipeline_events system table

A per-pipeline event log holds events for one pipeline only, which makes it awkward for monitoring dozens of pipelines. For that, Databricks recommends the pipeline_events system table, currently in Beta. It holds events from every pipeline in every workspace of a region. To query it, a user must either be both a metastore admin and an account admin, or have USE and SELECT on the system schemas. The data is the same kind of event, but some column names and the JSON access syntax differ:

To see the syntax difference on one metric, here is how each source reads backlog_bytes from flow_progress events. First the per-pipeline event log, which uses the : operator:

Double(details :flow_progress.metrics.backlog_bytes) as backlog

Then the system table, which uses variant_get on the same details path (and filters on event_time rather than timestamp):

variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT')

Databricks provides a dashboard built on this table that tracks updates, throughput, backlog, data quality and errors. Its queries show the health checks you would expect. One returns the most recent ERROR-level event for each pipeline. One finds expectations that failed records in the last day. One, shown below, counts errors per pipeline per hour so spikes stand out:

SELECT
  pipeline_id,
  date_trunc('HOUR', event_time) AS hour,
  count(*) AS num_errors
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  level = 'ERROR'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
GROUP BY
  pipeline_id,
  date_trunc('HOUR', event_time)
ORDER BY
  hour DESC

The dashboard's backlog query treats a flow whose status is COMPLETED as fully caught up and reports its backlog as 0. Its failed-expectations query groups by the dataset field inside each expectation entry and falls back to origin.dataset_name only when needed, because the expectation's own field is filled in even when origin.dataset_name is empty.

Sources4

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.Every expectation records passed and failed counts, so a missing count means the expectation never ran.Why is that wrong?

   An expectation with the fail action stops the update when it meets an invalid record, so no metrics are recorded for it. Metrics can also be missing for some datasets.

   Covered in Expectation metrics in flow_progress events

2. 2.Data quality results have their own event type in the event log.Why is that wrong?

   Expectation and dropped-record metrics are inside the details of flow_progress events, so a data quality query has to filter on event_type = 'flow_progress'.

   Covered in Expectation metrics in flow_progress events

Practise it for real

Query a pipeline's event log and report expectation metrics for its latest update

1. 1.Run SELECT * FROM event_log(<pipelineId>); as the pipeline's run-as user, replacing <pipelineId> with your pipeline's ID.

   Why: By default the event log is hidden, and only the run-as user can read it through event_log().

   You should see: Rows with id, timestamp, origin, level, event_type and a JSON details column.

2. 2.Create the event_log_raw view over the event log, using event_log(<pipelineId>) or the published table name.

   Why: The docs' monitoring queries all read from this view, and Databricks recommends granting access through a view rather than the table.

   You should see: A view that returns the same rows as the event log.

3. 3.Query event_log_raw for event_type = 'flow_progress' in the latest update and explode details:flow_progress:data_quality:expectations.

   Why: Per-expectation passed and failed counts are stored in that array on flow_progress events.

   You should see: One row per dataset and expectation with passing_records and failing_records. Expectations using the fail action will not appear.

4. 4.Query event_log_raw for flow_progress events and extract details:flow_progress.metrics.backlog_bytes.

   Why: backlog_bytes shows whether the flow is keeping up with its source.

   You should see: Backlog values over time for each flow, or none if the source type or runtime does not report backlog.