Quarantine bad rows instead of dropping them

Let’s take a stream of raw order rows where some fail your expectations — a null order_id, a negative amount — and instead of throwing those rows away, keep them. Route each failed row into a dead-letter table you can query, fix in place, and recycle back into the pipeline on the next run. The clean rows never stop flowing.

The Get Started course drops bad rows with mode: dqe: one @dp.expect_all_or_drop decorator, and a row that fails is gone. That is the right default when bad data is noise. But when a failed row is a customer order you cannot silently lose — because someone has to investigate it, backfill it, and get it into the table — dropping is the wrong verb. You want to quarantine it.

You could hand-write the machinery for that: a foreachBatch sink that MERGEs failed rows into a Delta table on a deterministic key so retries do not double-insert; an inverse filter that selects exactly the rows that failed any rule; a Change Data Feed reader that notices when an operator marks a row fixed; a windowed dedup so a row edited twice is ingested once; and a VARIANT-to-columns reconstruction that re-validates recycled rows before they rejoin the stream. Or you declare mode: quarantine with a two-field quarantine block, and let Lakehouse Plumber write all of it. That is the idea on every page: declare your ETL, don’t hand-write it.

Before you start

You need a Lakehouse Plumber project (see the Get Started course for lhp init) and a streaming source of rows — quarantine generates streaming code, so the data_quality action runs with readMode: stream. This guide reuses the orders ingest from the course: an Auto Loader load that streams raw CSV into v_orders_raw.

Quarantine also depends on two tables that you create yourself, not Lakehouse Plumber. The generated code reads and writes them by name; it does not run their DDL:

• The DLQ table (the inbox) named in dlq_table. It must have Change Data Feed and row tracking enabled — delta.enableChangeDataFeed = 'true' — or the recycle stream fails at run time.

• The outbox table, whose name Lakehouse Plumber derives by appending _outbox to dlq_table. No YAML configures it.

The exact column schema for both tables, and the one-time CREATE TABLE DDL, live in the Quarantine reference. Create them once per catalog and schema before the pipeline runs.

Declare the quarantine

Take the data_quality action and change one field. Where drop mode sets mode: dqe (or omits mode entirely), quarantine mode sets mode: quarantine and adds a quarantine block. That block has exactly two required fields — dlq_table and source_table:

pipelines/bronze_orders.yaml

# Quality & Ops guide: quarantine mode.
# Action chain: load (cloudFiles CSV) -> transform:data_quality (mode: quarantine) -> write.
# Failing rows are routed to a dead-letter table instead of being dropped.
pipeline: bronze_orders
flowgroup: orders_quarantine

actions:
  # 1. Load: stream CSV order files from the landing volume with Auto Loader.
  - name: load_orders_csv
    type: load
    source:
      type: cloudfiles
      path: "${landing_path}/orders/*.csv"
      format: csv
    target: v_orders_raw

  # 2. Transform (data quality, quarantine): rows that fail an expectation are
  #    routed to the DLQ table; clean rows flow on. Fixed DLQ rows are recycled
  #    back into the output on the next run.
  - name: quarantine_orders
    type: transform
    transform_type: data_quality
    source: v_orders_raw
    expectations_file: expectations/orders_quality.yaml
    mode: quarantine
    quarantine:
      dlq_table: "${catalog}.${bronze_schema}.orders_dlq"
      source_table: "${catalog}.${bronze_schema}.orders"
    target: v_orders_validated

  # 3. Write: append the validated rows (clean + recycled) into the bronze table.
  - name: write_orders_bronze
    type: write
    source: v_orders_validated
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${bronze_schema}"
      table: orders

The two fields carry the whole contract:

• dlq_table — the fully-qualified dead-letter table failed rows are written to. Here ${catalog}.${bronze_schema}.orders_dlq, which resolves per environment from substitutions/dev.yaml.

• source_table — the fully-qualified name this flowgroup tags its rows with in the DLQ. It is written into the _dlq_source_table column and used to filter the recycle stream, so one shared DLQ table can serve many flowgroups. Point it at the table the write action targets: ${catalog}.${bronze_schema}.orders.

The expectations file is the same YAML you would write for drop mode:

expectations/orders_quality.yaml

# Data-quality expectations for the orders flowgroup.
# In quarantine mode every rule is coerced to `drop`: a failing row is kept out
# of the clean stream AND written to the dead-letter table for later inspection.
order_id IS NOT NULL:
  action: drop
  name: valid_order_id
amount >= 0:
  action: drop
  name: non_negative_amount

Important

In quarantine mode every expectation is coerced to drop, whatever each rule’s action says. A row that fails any rule is held out of the clean stream and routed to the DLQ. If a rule is written as warn or fail, lhp validate emits a warning telling you the action was coerced — the per-rule warn and fail semantics do not apply here. Both rules above are already drop, so there is nothing to coerce.

Generate the pipeline

Validate first, then generate:

$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze_orders  0 files
✓ validate (0.33s)
1 validated · 0.3s

$ lhp generate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze_orders  1 file
✓ generate (0.33s)
✓ format (0.06s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.4s

validate enforces the symmetry between the two sides: mode: quarantine requires the quarantine block, and a quarantine block without mode: quarantine is rejected. It also fails an empty expectations file, because with no rules the inverse filter that selects failed rows would be meaningless.

Read the dead-letter subsystem

Open generated/dev/bronze_orders/orders_quarantine.py. This is the entire output — nothing is hidden behind a runtime. The load view and the write target at the ends are ordinary; everything between them is the dead-letter subsystem that the one mode: quarantine switch generated:

generated/dev/bronze_orders/orders_quarantine.py

# Generated by LakehousePlumber
# Pipeline: bronze_orders
# FlowGroup: orders_quarantine

from pyspark.sql import functions as F
from pyspark.sql.types import MapType, StringType
from pyspark.sql.window import Window
from pyspark import pipelines as dp
from delta.tables import DeltaTable

# Pipeline Configuration
PIPELINE_ID = "bronze_orders"
FLOWGROUP_ID = "orders_quarantine"


# ============================================================================
# SOURCE VIEWS
# ============================================================================


@dp.temporary_view()
def v_orders_raw():
    """Load data from csv files at /Volumes/dev_catalog/landing/files/orders/*.csv"""
    df = (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .load("/Volumes/dev_catalog/landing/files/orders/*.csv")
    )

    return df


# ============================================================================
# DATA QUALITY & QUARANTINE
# ============================================================================


# ----------------------------------------------------------------------------
# Quarantine: v_orders_raw → v_orders_validated
# ----------------------------------------------------------------------------

# --- Rules & constants ---

_EXPECTATIONS_v_orders_raw = {
    "valid_order_id": "order_id IS NOT NULL",
    "non_negative_amount": "amount >= 0",
}

_INVERSE_FILTER_v_orders_raw = "NOT ((order_id IS NOT NULL) AND (amount >= 0))"

_FAILED_RULE_EXPRS_v_orders_raw = [
    F.when(
        ~F.expr("order_id IS NOT NULL"),
        F.struct(
            F.lit("valid_order_id").alias("name"),
            F.lit("order_id IS NOT NULL").alias("rule"),
        ),
    ),
    F.when(
        ~F.expr("amount >= 0"),
        F.struct(
            F.lit("non_negative_amount").alias("name"),
            F.lit("amount >= 0").alias("rule"),
        ),
    ),
]

DLQ_TABLE_v_orders_raw = "dev_catalog.bronze.orders_dlq"
DLQ_OUTBOX_TABLE_v_orders_raw = "dev_catalog.bronze.orders_dlq_outbox"
SOURCE_TABLE_v_orders_raw = "dev_catalog.bronze.orders"

_HASH_EXCLUDE_COLS_v_orders_raw = [
    "_flowgroup_name",
    "_ingestion_timestamp",
    "_pipeline_name",
    "_pipeline_run_id",
    "_source_file",
]

_EXPECTATIONS_RECYCLED_v_orders_raw = {
    "valid_order_id": "order_id IS NOT NULL",
    "non_negative_amount": "amount >= 0",
}


# --- Clean path (provides DQ metrics in event log) ---
@dp.temporary_view()
@dp.expect_all_or_drop(_EXPECTATIONS_v_orders_raw)
def _clean_v_orders_raw():
    """Data quality checks for v_orders_raw — clean records"""
    return spark.readStream.table("v_orders_raw")


# --- Quarantine path (DLQ sink + routing) ---
@dp.foreach_batch_sink(name="dlq_sink_v_orders_raw")
def dlq_sink_v_orders_raw(batch_df, batch_id):
    """Write quarantined rows to DLQ table"""
    if batch_df.isEmpty():
        return

    spark = batch_df.sparkSession
    data_cols = [c for c in batch_df.columns if not c.startswith("_dlq")]
    _exclude = set(_HASH_EXCLUDE_COLS_v_orders_raw)
    hash_cols = [c for c in data_cols if c not in _exclude]

    batch = batch_df.withColumns(
        {
            "_row_json": F.to_json(F.struct(*hash_cols)),
            "_dlq_sk": F.xxhash64(
                F.lit(SOURCE_TABLE_v_orders_raw), F.to_json(F.struct(*hash_cols))
            ).cast("string"),
        }
    )

    if "_rescued_data" in batch_df.columns:
        _map_type = MapType(StringType(), StringType())
        variant_cols = [c for c in data_cols if c != "_rescued_data"]
        batch = (
            batch.withColumns(
                {
                    "_variant_json": F.when(
                        F.col("_rescued_data").isNotNull(),
                        F.to_json(
                            F.map_zip_with(
                                F.coalesce(
                                    F.from_json(
                                        F.to_json(F.struct(*variant_cols)), _map_type
                                    ),
                                    F.create_map(),
                                ),
                                F.map_filter(
                                    F.coalesce(
                                        F.from_json(F.col("_rescued_data"), _map_type),
                                        F.create_map(),
                                    ),
                                    lambda k, v: k != "_file_path",
                                ),
                                lambda k, v1, v2: F.coalesce(v2, v1),
                            )
                        ),
                    ).otherwise(F.to_json(F.struct(*variant_cols))),
                }
            )
            .withColumns(
                {
                    "_row_data": F.parse_json(F.col("_variant_json")),
                }
            )
            .withColumnRenamed("_rescued_data", "_dlq_rescued_data")
        )
    else:
        batch = batch.withColumns(
            {
                "_row_data": F.parse_json(F.to_json(F.struct(*data_cols))),
                "_dlq_rescued_data": F.lit(None).cast("string"),
            }
        )

    batch = batch.withColumns(
        {
            "_dlq_source_table": F.lit(SOURCE_TABLE_v_orders_raw),
            "_dlq_status": F.lit("quarantined"),
            "_dlq_timestamp": F.current_timestamp(),
        }
    ).select(
        "_dlq_sk",
        "_dlq_source_table",
        "_dlq_status",
        "_dlq_timestamp",
        "_dlq_failed_rules",
        "_dlq_rescued_data",
        "_row_data",
    )

    dlq = DeltaTable.forName(spark, DLQ_TABLE_v_orders_raw)
    (
        dlq.alias("dlq")
        .merge(batch.alias("new"), "dlq._dlq_sk = new._dlq_sk")
        .whenNotMatchedInsertAll()
        .execute()
    )
    return


@dp.append_flow(target="dlq_sink_v_orders_raw", name="quarantine_flow_v_orders_raw")
def quarantine_flow_v_orders_raw():
    """Route failed rows to DLQ"""
    return (
        spark.readStream.table("v_orders_raw")
        .filter(_INVERSE_FILTER_v_orders_raw)
        .withColumn(
            "_dlq_failed_rules",
            F.array_compact(F.array(*_FAILED_RULE_EXPRS_v_orders_raw)),
        )
    )


# --- Recycle path (dedup inbox → outbox) ---
@dp.foreach_batch_sink(name="recycle_sink_v_orders_raw")
def recycle_sink_v_orders_raw(batch_df, batch_id):
    """Deduplicate fixed DLQ rows and write to outbox"""
    if batch_df.isEmpty():
        return

    spark = batch_df.sparkSession
    w = Window.partitionBy("_dlq_sk").orderBy(F.desc("_commit_version"))
    deduped = (
        batch_df.withColumn("_rn", F.row_number().over(w))
        .filter("_rn = 1")
        .drop("_rn")
        .select(
            "_dlq_sk",
            "_dlq_source_table",
            "_row_data",
            F.current_timestamp().alias("_dlq_recycled_at"),
        )
    )

    outbox = DeltaTable.forName(spark, DLQ_OUTBOX_TABLE_v_orders_raw)
    (
        outbox.alias("out")
        .merge(deduped.alias("new"), "out._dlq_sk = new._dlq_sk")
        .whenNotMatchedInsertAll()
        .execute()
    )


@dp.append_flow(target="recycle_sink_v_orders_raw", name="recycle_flow_v_orders_raw")
def recycle_flow_v_orders_raw():
    """Read fixed rows from DLQ inbox via CDF"""
    return (
        spark.readStream.option("readChangeFeed", "true")
        .table(DLQ_TABLE_v_orders_raw)
        .filter(
            "_dlq_status = 'fixed' "
            "AND _change_type IN ('insert', 'update_postimage') "
            f"AND _dlq_source_table = '{SOURCE_TABLE_v_orders_raw}'"
        )
    )


# --- Recycled path (outbox → validated recycled view) ---
@dp.temporary_view()
@dp.expect_all_or_drop(_EXPECTATIONS_RECYCLED_v_orders_raw)
def _recycled_v_orders_raw():
    """Validate recycled rows from DLQ outbox"""
    clean = spark.readStream.table("_clean_v_orders_raw")
    return (
        spark.readStream.option("skipChangeCommits", "true")
        .table(DLQ_OUTBOX_TABLE_v_orders_raw)
        .filter(f"_dlq_source_table = '{SOURCE_TABLE_v_orders_raw}'")
        .select(
            [
                F.try_variant_get(
                    F.col("_row_data"), f"$.{field.name}", field.dataType.simpleString()
                ).alias(field.name)
                for field in clean.schema.fields
            ]
        )
    )


# --- Validated output (clean + recycled) ---
@dp.temporary_view()
def v_orders_validated():
    """Data quality checks for v_orders_raw — clean + recycled records"""
    clean = spark.readStream.table("_clean_v_orders_raw")
    recycled = spark.readStream.table("_recycled_v_orders_raw")

    df = clean.union(recycled)

    return df


# ============================================================================
# TARGET TABLES
# ============================================================================

# Create the streaming table
dp.create_streaming_table(
    name="dev_catalog.bronze.orders", comment="Streaming table: orders"
)


# Define append flow(s)
@dp.append_flow(
    target="dev_catalog.bronze.orders",
    name="f_orders_bronze",
    comment="Append flow to dev_catalog.bronze.orders",
)
def f_orders_bronze():
    """Append flow to dev_catalog.bronze.orders"""
    # Streaming flow
    df = spark.readStream.table("v_orders_validated")

    return df

Read it top to bottom and the inbox/outbox design is all there:

• Rules and constants. _EXPECTATIONS_v_orders_raw holds every rule as a drop. _INVERSE_FILTER_v_orders_raw is the negation — NOT ((order_id IS NOT NULL) AND (amount >= 0)) — which matches exactly the rows that failed at least one rule. _FAILED_RULE_EXPRS_v_orders_raw builds, per row, the {name, rule} structs recording which rules it broke. The resolved table names sit in DLQ_TABLE_, DLQ_OUTBOX_TABLE_ (the auto-derived orders_dlq_outbox), and SOURCE_TABLE_ constants.

• The clean path. _clean_v_orders_raw applies @dp.expect_all_or_drop to the source view. Passing rows flow on; this is also what surfaces the pass/fail metrics in the Databricks event log.

• The quarantine path. quarantine_flow_v_orders_raw reads the same source, applies the inverse filter to select only failed rows, and annotates each with _dlq_failed_rules. Its target, dlq_sink_v_orders_raw, is a @dp.foreach_batch_sink that hashes each row to a deterministic _dlq_sk (xxhash64 of the source table plus the row JSON) and MERGEs it into the DLQ, so a retry never double-inserts the same row.

• The recycle path. recycle_flow_v_orders_raw reads the DLQ through Change Data Feed, filtered to rows an operator has marked _dlq_status = 'fixed'. recycle_sink_v_orders_raw dedups them by keeping the latest _commit_version per _dlq_sk and merges them into the outbox — the permanent “already recycled” ledger that stops a re-edited row being ingested twice.

• The final view. _recycled_v_orders_raw reconstructs each recycled row from the VARIANT _row_data column with try_variant_get, re-runs the expectations on it, and v_orders_validated — the view your write reads — UNIONs the clean stream with the validated recycled stream.

Note

The recycle loop has a one-run lag by design: a row you mark fixed during run N is read from the CDF, deduped into the outbox, and re-joined to the output on run N+1. This follows from the streaming checkpoint model, and it is why quarantine runs in triggered (non-continuous) pipelines. The full list of limitations is in the Quarantine reference.

To operate the queue — query quarantined rows, see which rule each one broke, and mark a corrected row fixed so it recycles — you work directly against the DLQ table in SQL. Those recipes, with the _dlq_sk / _dlq_status / _dlq_failed_rules column details, are in the Quarantine reference; this guide stays on turning the mode on and reading what it wrote.

What you just did

You changed one field — mode: dqe became mode: quarantine — and added a four-line quarantine block. In drop mode that same data_quality action compiles to a single @dp.expect_all_or_drop decorator on one view. In quarantine mode it expands into a full dead-letter subsystem: 39 lines of YAML became a 296-line Lakeflow file, and roughly 230 of those lines are DLQ plumbing you did not write — the foreach_batch_sink MERGE, the inverse-filter routing flow, the CDF recycle reader, the windowed outbox dedup, and the VARIANT reconstruction that re-validates recycled rows.

Not one MERGE, foreachBatch, readChangeFeed, or xxhash64 came from you. You declared where failed rows go — dlq_table and source_table — and Lakehouse Plumber owns the recycling machinery. The generated Python is still code you own: version it, diff it, open it in the Databricks editor like any other file.

What’s next

• Create the DLQ tables and operate the queue. The Quarantine reference has the one-time CREATE TABLE DDL (with Change Data Feed and row tracking), the full column schema for the inbox and outbox, and the SQL recipes for querying and marking rows fixed.

• Stay with drop when you don’t need the row back. If a failed row is noise rather than a record to recover, mode: dqe and a single @dp.expect_all_or_drop is the lighter choice — covered in the Get Started course and the data-quality guide.

• Share one DLQ across flowgroups. Because every row is tagged with its source_table, several flowgroups can point dlq_table at the same dead-letter table and still recycle independently. Set the shared dlq_table in a preset and let each flowgroup supply its own source_table.