Declarative Pipeline Expectations - Quarantining Bad Records

[ChristianRRL]

Hi there,

I'm trying to follow Expectation recommendations and advanced patterns | Quarantine invalid records, but this doesn't really make sense to me.

Following the described pattern almost exactly in my case, I see that there are two TEMPORARY LIVE VIEWs generated (one for valid records and another for quarantined records).

Sure enough, I see there are two temp views generated, but I don't understand what their practical use case is here. The TEMPORARY STREAMING TABLE does show two records out of 6.9k that did not pass the expectation (and are showing as warning), but I don't see specifically which records those are. And when checking the TEMPORARY LIVE VIEWs, they don't really give me any information at all.

Can I get some help to understand this pattern better?

[veeramadaan]

The practical purpose of this pattern is to create two internal processing branches.

trips_data_quarantine classifies each record by adding is_quarantined and applies the expectations so that the pipeline records data-quality metrics. The two temporary views are simply reusable filters:

• valid_trips_data contains records where is_quarantined = FALSE and is intended to feed the normal downstream flow.

• invalid_trips_data contains records where is_quarantined = TRUE and is intended to feed a remediation, audit, or replay flow.

This is useful when the two branches have different downstream consumers. For example, valid trips could be written to a curated table, while invalid trips could be written to a quarantine table for investigation and later reprocessing.

[LiresaFerizaj]

Hi! The confusion here is understandable, because the docs example stops one step short of being useful on its own.

Why the views show nothing

Temporary views in a pipeline aren't materialized or published to Unity Catalog. They're logical building blocks meant to be read by other datasets within the same pipeline. If nothing downstream consumes them, they're never computed, so there are no metrics and nothing to query. Similarly, the temporary streaming table isn't published, and the expectation metrics in the UI only store counts in the event log, not the actual failing rows. That's why you can see "2 failed" but not which two.

How to actually see the quarantined records

Replace the two temporary views with published tables that read from the temp table and filter on the flag, for example:

CREATE OR REFRESH STREAMING TABLE quarantine_eleven_mile_outages AS
SELECT * FROM STREAM(temp_eleven_mile_outages) WHERE is_quarantined = TRUE;

Do the same with is_quarantined = FALSE for your silver table. Both will then appear in Unity Catalog, show row counts in the pipeline graph, and can be queried directly to review and fix the bad records. Alternatively, if you just want a quick look, you can drop the TEMPORARY keyword from the flagged table and query WHERE is_quarantined = TRUE.

One thing to double-check

Make sure your is_quarantined expression is the exact negation of your valid_id expectation. The expectation and the flag are evaluated independently, so if the logic drifts apart, the counts in the UI won't match what lands in the quarantine table.

Hope that helps!

[VladTamas]

Short answer: the two temporary views are empty because nothing reads them. Views in a pipeline are only computed when another dataset queries them, and the temporary streaming table isn't published to Unity Catalog. To see the actual bad rows, read the flagged table into a published table. To see the counts per expectation, query the pipeline event log.

1. What the pattern does

The docs pattern [1] has three parts:

• A temporary streaming table that adds an is_quarantined flag and applies the expectations, so the event log records data quality metrics.
• Two temporary views that split the rows on that flag, one for valid rows and one for invalid rows.
• The views are meant to feed downstream datasets in the same pipeline. They aren't meant to be queried directly.

2. Why the temporary views show nothing

Views in a pipeline are calculated only when queried, so a view that no downstream dataset reads may have no data quality metrics at all [2]. The temporary (now called private) table isn't published to the metastore and can't be accessed outside the pipeline [3]. The expectation UI shows counts ("2 failed") but not the failing rows.

3. See which rows were quarantined

Add published tables that read from the flagged table:

CREATE OR REFRESH STREAMING TABLE trips_quarantine AS
SELECT * FROM STREAM(trips_data_quarantine) WHERE is_quarantined = TRUE;

CREATE OR REFRESH STREAMING TABLE trips_clean AS
SELECT * FROM STREAM(trips_data_quarantine) WHERE is_quarantined = FALSE;

Both tables then show up in Unity Catalog, and you can query trips_quarantine to inspect and fix the rows.

4. See the counts per expectation

The pass and fail counts are in flow_progress events in the event log. By default, the pipeline's run-as user can read them with event_log() [4]:

SELECT timestamp, origin.flow_name, e.name AS expectation, e.passed_records, e.failed_records
FROM event_log('<pipeline-id>'),
  LATERAL explode(from_json(details:flow_progress.data_quality.expectations,
    'array<struct<name:string,dataset:string,passed_records:bigint,failed_records:bigint>>')) AS t(e)
WHERE event_type = 'flow_progress'
  AND details:flow_progress.data_quality.expectations IS NOT NULL
ORDER BY timestamp DESC;

In my test on samples.nyctaxi.trips, this returned one row per expectation (for example valid_trip_distance: 21,856 passed, 76 failed), and the quarantine table held exactly the 85 rows that failed.

What I'd do

Keep the pattern, but replace the two temporary views with published streaming tables like the ones in step 3. Make sure is_quarantined is exactly NOT(<all expectation rules>), so the quarantine table matches the failed counts in the event log. The Python version of the docs pattern builds the flag from the same rules dict for this reason [1].

References

[1] Expectation recommendations and advanced patterns | Databricks on AWS — docs.databricks.com/aws/ldp/expectation-patterns
[2] Manage data quality with pipeline expectations | Databricks on AWS — docs.databricks.com/aws/ldp/expectations
[3] table (Python language reference) | Databricks on AWS — docs.databricks.com/aws/ldp/developer/ldp-python-ref-table
[4] Pipeline event log | Databricks on AWS — docs.databricks.com/aws/ldp/monitor-event-logs

[data_pulse]

Adding further to above good replies, one thing I would like to add is that a quarantine table is much more useful if it stores why each row was quarantined, not just is_quarantined = true.

The expectation UI/event log is mainly for counts and trends, it is not a row-level error store and doesn't give full picture.

For multiple DQ rules, consider adding a failed_rules column to the quarantined dataset, for example:

rules = {
    "valid_pickup_zip": "pickup_zip IS NOT NULL",
    "valid_dropoff_zip": "dropoff_zip IS NOT NULL"
}

quarantine_condition = " OR ".join(
    f"NOT ({expr})"
    for expr in rules.values()
)

failed_rule_exprs = [
    F.when(~F.expr(condition), F.lit(rule_name))
    for rule_name, condition in rules.items()
]
df = (
    source_df
    .withColumn(
        "failed_rules",
        F.array_compact(F.array(*failed_rule_exprs))
    )
)

Then quarantine and clean rows are simply :
quarantined_df = df.filter(F.size("failed_rules") > 0)
clean_df = df.filter(F.size("failed_rules") == 0)

That gives you both:

• expectation metrics for monitoring
• row level failure reasons for investigation/reprocessing

One other thing for production is to keep the expectation rules and quarantine logic driven from the same rule set where possible. They could match today, but defining them separately makes it easy for them to drift over time when someone updates one and forgets the other. It's better to maintain single version of truth for the rules defined.