Migrating Legacy ETL Pipelines to Databricks: Architecture and Validation

Most legacy ETL migrations do not fail on code conversion. They fail at the end, when a business owner asks whether the new numbers can be trusted and the honest answer is “mostly”. Years of Talend, Informatica or SSIS jobs encode business rules where a converter never looks: lookup policies, implicit casts, the order in which rows happened to arrive, the way a database compares strings.

This article covers how to triage a legacy estate, the target architecture on Databricks, the ETL semantics that need deliberate translation, how to sequence waves with dual-running and rollback, and, at the centre, a reconciliation framework with working PySpark and SQL that produces evidence a business owner can sign.

Inventory before you convert

Start with facts, not with a conversion tool. For every job, capture its dependencies, sources and targets, components, schedule, recent run history and, above all, who consumes its output. Most ETL tools export their repository as XML or JSON, so the structure is scriptable; run history and consumers come from scheduler logs, database audit data and report owners.

Two findings repay the effort. Usage evidence, because the cheapest job to migrate is the one you retire. And lineage between jobs, because it decides which jobs must move together, which is what a migration wave really is.

Lakebridge, a Databricks Labs project, includes an Analyzer that scans exported ETL and SQL metadata for complexity, components and cross-job dependencies, alongside transpilers and a reconciliation module. The project is provided as-is, without Databricks support SLAs, so treat its output as an input to the inventory rather than the inventory itself.

Classify every job

Give each job exactly one disposition. It drives effort, target design and, critically, how the job will be validated.

Score each job on business criticality, complexity (component count, custom code, stateful logic) and coupling, then build waves from dependency clusters rather than from folders in the ETL repository.

Target architecture on the lakehouse

Layered tables with explicit contracts

Bronze holds data as received, with ingestion metadata. Silver reproduces legacy semantics deliberately: types, trimming, null rules, deduplication, conformed keys. Gold holds the business-facing outputs, each corresponding to one legacy output so that it has a reconciliation pair; legacy staging tables need no equivalent. For the catalog layout underneath, see the companion article on designing enterprise data governance with Unity Catalog.

Ingestion

Auto Loader (the cloudFiles source) replaces most “poll a directory, load, archive” patterns, discovering files incrementally and processing each exactly once into Delta. Two defaults matter. For CSV and JSON it infers every column as a string unless you set cloudFiles.inferColumnTypes or supply cloudFiles.schemaHints. Its default evolution mode, addNewColumns, stops the stream when a new column appears and picks up the new schema on restart, and values that do not fit land in _rescued_data instead of disappearing, which is what you want while discovering what legacy feeds really contain. For databases and SaaS applications, Lakeflow Connect offers managed connectors, including change data capture from SQL Server; release stages vary by connector, so check the current documentation.

Declarative pipelines or jobs

Lakeflow pipelines (Lakeflow Spark Declarative Pipelines, formerly Delta Live Tables) define streaming tables and materialized views, resolve the dependency graph, retry transient failures and enforce expectations with warn, drop or fail actions. Their AUTO CDC flows, which replace the APPLY CHANGES APIs, implement SCD Type 1 and Type 2 and handle out-of-order events.

Lakeflow Jobs orchestrates the rest, with task dependencies, If/else and For each tasks, and file-arrival and table-update triggers. Repairing a failed run re-runs only the unsuccessful tasks and their dependents, but the platform does not make a task idempotent; your write pattern must. Deploy both with Declarative Automation Bundles (formerly Databricks Asset Bundles), and replace every scheduler time offset (“B starts at 02:30 because A is usually done by 02:15”) with an explicit dependency.

Semantics that do not map one-to-one

A transpiler converts syntax. These behaviours need decisions, recorded per job or per column.

Row-by-row and stateful logic

Expressions that carry a value over from the previous row, script components and custom routines assume rows arrive in a stable order; Spark guarantees no order without an ORDER BY. Translate stateful logic into window functions with an explicit, deterministic ordering key. If the legacy job relied on read order and no such key exists, its output is itself non-deterministic, and the owner must agree that before you try to match it.

Lookups

Lookup components resolve duplicates silently, and differently. Talend’s tMap defaults to the Unique match model, which returns the last matching lookup row. The SSIS Lookup transformation returns the first match; in full cache mode it matches case- and trailing-space-sensitively, while in no-cache and partial-cache modes it follows the source database’s comparison rules. A Spark join returns every match and multiplies rows. For each lookup, assert that the key is unique in the reference data or reproduce the policy with row_number() over a deterministic order, and turn no-match outputs into a left join with a flag or reject table, never an inner join that drops rows.

Surrogate keys

Delta identity columns (GENERATED ALWAYS AS IDENTITY) produce unique values that are not guaranteed to be contiguous, and declaring one disables concurrent transactions on the table. New surrogate keys will not equal the legacy ones, so reconcile dimensions on business keys, and facts by translating their foreign keys back to business keys. If consumers depend on legacy key values, load history into a GENERATED BY DEFAULT AS IDENTITY column, which accepts explicit values, then run ALTER TABLE ... ALTER COLUMN ... SYNC IDENTITY.

Slowly changing dimensions

SCD wizards and multi-step update-then-insert flows become one set-based MERGE. The pattern below stages each change twice: keyed on the business key, to close the current version, and with a NULL key that can never match, so that the new version is inserted.

CREATE TABLE IF NOT EXISTS main.sales.dim_customer (
  customer_sk BIGINT GENERATED ALWAYS AS IDENTITY,
  customer_id STRING NOT NULL,  -- business key
  name        STRING,
  segment     STRING,
  city        STRING,
  valid_from  TIMESTAMP NOT NULL,
  valid_to    TIMESTAMP,        -- NULL while current
  is_current  BOOLEAN NOT NULL
);

WITH latest AS (
  -- One change per business key per batch; otherwise MERGE fails on multiple matches.
  SELECT customer_id, name, segment, city, changed_at
  FROM main.staging.customer_changes
  QUALIFY row_number() OVER (PARTITION BY customer_id ORDER BY changed_at DESC) = 1
),
staged AS (
  -- Every change, keyed on the business key: closes the current version of a
  -- changed customer, or falls through to INSERT for a brand-new customer.
  SELECT customer_id AS merge_key, * FROM latest
  UNION ALL
  -- Changed customers once more with a NULL key. These never match, so they are
  -- inserted as the new current version.
  SELECT NULL AS merge_key, s.*
  FROM latest AS s
  JOIN main.sales.dim_customer AS d
    ON d.customer_id = s.customer_id AND d.is_current
  WHERE NOT (d.name <=> s.name AND d.segment <=> s.segment AND d.city <=> s.city)
)
MERGE INTO main.sales.dim_customer AS d
USING staged AS s
ON d.customer_id = s.merge_key AND d.is_current
WHEN MATCHED
  AND NOT (d.name <=> s.name AND d.segment <=> s.segment AND d.city <=> s.city)
  THEN UPDATE SET valid_to = s.changed_at, is_current = false
WHEN NOT MATCHED THEN
  INSERT (customer_id, name, segment, city, valid_from, valid_to, is_current)
  VALUES (s.customer_id, s.name, s.segment, s.city, s.changed_at, NULL, true);

Three details carry the correctness. The batch is reduced to one change per key, because MERGE fails when several source rows match the same target row. Change detection uses the null-safe <=>, so a value changing to or from NULL counts as a change. And re-running a batch changes nothing, which makes retries and repair runs safe.

The pattern assumes batches arrive in order and ignores deletes. If either fails, AUTO CDC ... STORED AS SCD TYPE 2 with SEQUENCE BY handles out-of-order events, but its generated __START_AT and __END_AT columns must then be mapped onto the legacy validity columns, including whether the open row ends in NULL or a sentinel such as 9999-12-31.

Reject flows

Reject links and error outputs become tables. In pipelines, use expectations (EXPECT ... ON VIOLATION DROP ROW or FAIL UPDATE) and a quarantine table for rows you want to inspect; in job tasks, split explicitly and write rejects with a reason code. Reconcile reject counts as a metric of their own: a migrated job that rejects fewer rows than the legacy one is as suspicious as one that rejects more.

Implicit type coercion

ANSI mode is enabled by default in Databricks Runtime 17.0 and above, as in Apache Spark 4.0: an invalid cast raises an error instead of returning NULL, and integer overflow raises instead of wrapping. Legacy tools often coerced silently. Decide per column whether to fail loudly or to reproduce the leniency explicitly with try_cast, routing failures to the reject table. Decide, too, where rounding happens: per line and per total give different answers.

Timestamps need rules too. SQL Server’s datetime is rounded to increments of .000, .003 or .007 seconds, while Spark’s TIMESTAMP defaults to microsecond precision. Local wall-clock times stored without a zone should map to TIMESTAMP_NTZ or be converted explicitly, not left to the session time zone.

Nulls, empty strings and trailing spaces

Oracle treats an empty string as NULL; most engines do not. SQL Server’s = pads the shorter string with spaces, so 'abc' = 'abc ' is true; under Spark’s default UTF8_BINARY collation it is false, and fixed-width CHAR columns usually arrive padded. Normalise in silver with rtrim and an agreed empty-string rule, or use a collation with the RTRIM modifier (Databricks Runtime 16.2 and above) where you want the source behaviour. Either way, the reconciliation must apply the same rule.

Sequencing: waves, dual-run, cutover and rollback

Build waves from dependency clusters. Make the first wave low in business risk but representative: a pilot that exercises every pattern you will need (file and database ingestion, SCD, rejects, downstream extracts), not simply the easiest job.

During dual-run, both pipelines consume the same snapshot of inputs and their outputs are compared on every run. The shared snapshot is the point: if the legacy job reads a source at 01:00 and the new pipeline at 01:20, the dual-run becomes an exercise in explaining timing. Land legacy outputs in a reconciliation schema in Unity Catalog and compare there.

Cut over per consumer, not per pipeline, ideally through views that preserve legacy column names. Agree rollback triggers in advance and keep the legacy jobs runnable until sign-off. Delta’s RESTORE TABLE ... TO VERSION AS OF returns a table to its pre-cutover state, provided VACUUM has not removed the files that version needs.

The reconciliation framework

Reconciliation answers one question: for the same inputs, does the new pipeline produce the outputs the business expects, and is every difference explained?

Levels of evidence

1. Schema parity: names, types and nullability against the agreed mapping.
2. Counts: rows and rejects per business date or partition.
3. Aggregates: measure sums by business dimension, distinct keys, date ranges and an order-independent fingerprint.
4. Keyed diff: a column-level comparison joined on the business key.
5. Invariants, for redesigned jobs: totals agree with the ledger, no key has two current versions, every fact resolves to a dimension row.

Levels 1 to 3 are cheap enough for every load. Level 4 is the proof: run it on every dual-run load if you can afford it, and at least on every period close. Always compare canonical values (trimmed strings, the agreed empty-string rule, decimals at an agreed scale, timestamps in UTC), and keep those rules in version-controlled code: they are the contract with the data owner.

Keyed column-level diff

The function below joins a legacy output, landed in the reconciliation schema, to the migrated table on the business key and returns one row per discrepancy: a missing record, an unexpected record, or a column whose values differ beyond its rule.

from functools import reduce

from pyspark.sql import DataFrame, functions as F

KEYS = ["order_id"]
# One rule per column, agreed with the data owner and versioned with this code.
RULES = {
    "customer_id": "exact",
    "status": "rtrim",          # legacy CHAR(12) column is blank-padded
    "order_ts": "ts_ms:3",      # legacy DATETIME is accurate to about 3 ms
    "net_amount": "abs:0.005",  # DECIMAL money: half a cent either way
    "fx_rate": "rel:1e-9",      # DOUBLE: relative tolerance
}


def matches(col: str, rule: str):
    """Null-safe comparison of l.<col> and r.<col>. Never returns NULL."""
    l, r = F.col(f"l.{col}"), F.col(f"r.{col}")
    kind, _, arg = rule.partition(":")
    if kind == "exact":
        return l.eqNullSafe(r)
    if kind == "rtrim":
        return F.rtrim(l).eqNullSafe(F.rtrim(r))
    if kind == "ts_ms":
        within = F.abs(F.unix_micros(l) - F.unix_micros(r)) <= int(arg) * 1000
    elif kind == "abs":
        within = F.abs(l - r) <= float(arg)
    elif kind == "rel":
        within = F.abs(l - r) <= float(arg) * F.greatest(F.abs(l), F.abs(r))
    else:
        raise ValueError(f"unknown rule {rule!r} for column {col}")
    return (l.isNull() & r.isNull()) | F.coalesce(within, F.lit(False))


def assert_clean_keys(df: DataFrame, name: str) -> None:
    null_key = reduce(lambda a, b: a | b, [F.col(k).isNull() for k in KEYS])
    bad = df.groupBy(*KEYS).count().where((F.col("count") > 1) | null_key)
    if bad.limit(1).count():
        raise ValueError(f"{name}: null or duplicate keys; a keyed diff is meaningless")


def reconcile(legacy: DataFrame, migrated: DataFrame) -> DataFrame:
    """One row per discrepancy: a missing or unexpected record, or a column mismatch."""
    joined = legacy.alias("l").join(
        migrated.alias("r"),
        [F.col(f"l.{k}") == F.col(f"r.{k}") for k in KEYS],
        "full_outer",
    )
    record_key = F.to_json(F.struct(*[
        F.coalesce(F.col(f"l.{k}"), F.col(f"r.{k}")).alias(k) for k in KEYS
    ])).alias("record_key")
    in_l = F.col(f"l.{KEYS[0]}").isNotNull()
    in_r = F.col(f"r.{KEYS[0]}").isNotNull()
    none = F.lit(None).cast("string")

    orphans = joined.where(~(in_l & in_r)).select(
        record_key,
        F.when(in_l, "missing_in_migrated")
        .otherwise("unexpected_in_migrated").alias("status"),
        none.alias("column_name"), none.alias("legacy_value"),
        none.alias("migrated_value"),
    )
    diffs = F.array(*[
        F.when(~matches(c, rule), F.struct(
            F.lit(c).alias("column_name"),
            F.col(f"l.{c}").cast("string").alias("legacy_value"),
            F.col(f"r.{c}").cast("string").alias("migrated_value"),
        ))
        for c, rule in RULES.items()
    ])
    mismatches = (
        joined.where(in_l & in_r)
        .select(record_key, F.explode(F.filter(diffs, lambda d: d.isNotNull())).alias("d"))
        .select("record_key", F.lit("value_mismatch").alias("status"), "d.*")
    )
    return orphans.unionByName(mismatches)

A run for one business date:

run_id = "orders-2026-11-02-r1"
day = "business_date = '2026-11-02'"
legacy = spark.table("main.recon.legacy_orders").where(day)
migrated = spark.table("main.sales.orders").where(day)

assert_clean_keys(legacy, "legacy")
assert_clean_keys(migrated, "migrated")

(
    reconcile(legacy, migrated)
    .withColumn("run_id", F.lit(run_id))
    .withColumn("table_name", F.lit("main.sales.orders"))
    .write.mode("append")
    .saveAsTable("main.recon.diff_results")
)

Every comparison is null-safe and never yields NULL. That matters: a NULL predicate in a filter silently drops the row, and a reconciliation that drops rows reports success. One row per column difference also exposes the failure pattern: one column wrong on every row is a translation bug, while scattered differences usually point to input timing or a lookup policy.

For unit tests of translated logic on small fixtures, pyspark.testing.assertDataFrameEqual offers rtol and atol tolerances; it is a test assertion, not a full-volume reconciliation. Lakebridge’s Reconcile module, which compares schemas, row hashes and keyed data, is worth evaluating too; the principles hold whichever engine runs the comparison.

Aggregates and fingerprints

Aggregates catch volume and value problems at a fraction of the cost of a keyed diff, and they are the numbers business owners recognise.

WITH sides AS (
  SELECT 'legacy' AS side, business_date, order_id, customer_id, status, net_amount
  FROM main.recon.legacy_orders
  UNION ALL
  SELECT 'migrated', business_date, order_id, customer_id, status, net_amount
  FROM main.sales.orders
),
signed AS (
  SELECT
    side,
    business_date,
    CASE side WHEN 'migrated' THEN 1 ELSE -1 END AS sign,
    net_amount,
    -- DECIMAL(38,0): summing 64-bit hashes as BIGINT would overflow under ANSI mode
    CAST(xxhash64(order_id, customer_id, rtrim(status)) AS DECIMAL(38, 0)) AS row_hash
  FROM sides
)
SELECT
  business_date,
  count_if(side = 'legacy')   AS legacy_rows,
  count_if(side = 'migrated') AS migrated_rows,
  sum(sign * net_amount)      AS net_amount_delta,
  sum(sign * row_hash) = 0    AS fingerprint_match
FROM signed
GROUP BY business_date
ORDER BY business_date;

Two details are deliberate. Hashes are summed as DECIMAL(38, 0) because a BIGINT sum of 64-bit hashes overflows, and under ANSI mode overflow is an error. And the fingerprint is a screen, not proof: xxhash64 skips NULL arguments, so xxhash64(1, NULL) equals xxhash64(NULL, 1). A fingerprint mismatch says look closer; the keyed diff says where.

Tolerance rules

• Floating point: use a relative tolerance. Floating-point addition is not associative, so a distributed sum of DOUBLE values can differ in its last digits between engines and even between runs.
• Money: compare DECIMAL exactly. Needing a tolerance usually means rounding happens at a different step; fix the step, or document the reason.
• Timestamps: compare within a window matching the source precision, and test daylight-saving transitions wherever local times were converted.

Each tolerance is a decision with an owner. There is no global “close enough”.

Legitimate differences

Some differences are correct. The new pipeline may fix a legacy defect, a redesign may change a rule on purpose, and some legacy behaviour is non-deterministic, such as a lookup that returned an arbitrary match. Record each in an accepted-differences table with its scope (table, status, column, optionally a key), reason, approver and expiry. The number that must reach zero is the count of unexplained differences:

SELECT d.status, d.column_name, count(*) AS unexplained
FROM main.recon.diff_results AS d
LEFT ANTI JOIN main.recon.accepted_differences AS a
  ON  a.table_name = d.table_name
  AND a.status = d.status
  AND a.column_name <=> d.column_name
  AND (a.record_key IS NULL OR a.record_key = d.record_key)
  AND a.expires_on >= current_date()
WHERE d.run_id = :run_id
GROUP BY d.status, d.column_name
ORDER BY unexplained DESC;

Evidence for sign-off

For each table and run, keep the input snapshot identifier, the Delta versions of both compared tables (from DESCRIBE HISTORY), counts and aggregate deltas, the unexplained-difference count, accepted differences with their approvers, and the commit of the rule set used. Store it as Delta tables in the reconciliation schema, queryable and governed. Agree the sign-off criteria before dual-run begins; negotiated afterwards, every difference becomes a debate.

Decision checklist

• Every job has a disposition, an owner and a list of consumers; retirements are signed off.
• Every gold table maps to one legacy output or has an agreed set of invariants.
• Semantic decisions are recorded per column: trimming, empty strings, null keys, casts, rounding, time zones and lookup policies.
• Surrogate keys are reconciled through business keys.
• Dual-run inputs are snapshotted, and legacy outputs are landed in a governed reconciliation schema.
• Comparison rules and tolerances are versioned code with named owners.
• Exit criteria, rollback triggers and the legacy decommission date are agreed before cutover.

Migrations of this shape are the core of ByteLoom’s Enterprise Data Platform Modernization service, from inventory to reconciliation evidence. If you are planning one, discuss your project.

References

• Lakeflow pipeline concepts: flows, streaming tables and materialized views
• The AUTO CDC APIs: change data capture and SCD Type 2 with Lakeflow pipelines
• Auto Loader schema inference and evolution
• MERGE INTO reference for Databricks SQL
• ANSI compliance in Databricks Runtime
• Delta Lake generated and identity columns on Databricks
• Lakebridge Reconcile guide (Databricks Labs)
• SSIS Lookup transformation: matching and cache modes
• Talend tMap match models: Unique match, First match and All matches
• PySpark assertDataFrameEqual API reference