Migrating from SFTP to a roster API: dual-run and reconciliation

Part 7 of 7 — Roster API Series: migrating member-file ingestion from CSV over SFTP to a FHIR API.

Six posts built the target: a FHIR Group with explicit add/remove (Part 3), bulk export with deltas (Part 4), SMART auth (Part 5) and change notifications (Part 6). None of it matters if the cutover drops 3% of a cohort that drives billing and quality reporting.

So the last post is the boring, important one: how to run both pipes side by side, prove they agree, and only then turn SFTP off.

The phases

Phase 0  SFTP only ─────────────────────────────────────────────────────►
Phase 1  SFTP (source of truth)  +  API shadow (built from same CSV)
Phase 2  SFTP (source of truth)  +  API from upstream ops  →  reconcile daily
Phase 3  API (source of truth)   +  SFTP generated from API →  reconcile daily
Phase 4  API only; SFTP folder read-only, then decommissioned

Two principles:

1. Only one source of truth per phase. Dual-run means dual read, not dual authority. When they disagree, the phase table says who wins.
2. Phase 3 inverts the direction. Legacy consumers keep getting a CSV, but it’s now a projection of the API. That’s how you migrate producers and consumers on different schedules.

The reconciliation

Every day during phases 1–3, compare the CSV snapshot with the API export (the NDJSON from Part 4). DuckDB is perfect for this: it reads CSV and NDJSON natively, no loading step.

# reconcile.py
import duckdb

con = duckdb.connect()

con.execute("""
CREATE OR REPLACE VIEW csv_members AS
SELECT
    trim(mrn)                         AS mrn,
    upper(trim(last_name))            AS last_name,
    CAST(dob AS DATE)                 AS dob,
    trim(provider_npi)                AS provider_npi,
    CAST(period_start AS DATE)        AS period_start,
    CAST(period_end AS DATE)          AS period_end
FROM read_csv('sftp/roster_2026-10-31.csv', header = true, all_varchar = true);
""")

con.execute("""
CREATE OR REPLACE VIEW api_patients AS
SELECT
    id,
    list_filter(identifier, lambda x: x.system = 'https://ehr.example-clinic.org/fhir/sid/mrn')[1].value AS mrn,
    upper(name[1].family)  AS last_name,
    CAST(birthDate AS DATE) AS dob
FROM read_json('export/Patient.ndjson', format = 'newline_delimited');
""")

con.execute("""
CREATE OR REPLACE VIEW api_members AS
WITH g AS (
    SELECT unnest(member) AS m
    FROM read_json('export/Group.ndjson', format = 'newline_delimited')
)
SELECT
    split_part(m.entity.reference, '/', 2)                       AS patient_id,
    coalesce(m.inactive, false)                                  AS inactive,
    CAST(m.period.start AS DATE)                                 AS period_start,
    CAST(m.period."end" AS DATE)                                 AS period_end,
    list_filter(m.extension, lambda e: e.url LIKE '%ext-attributedProvider')[1]
        .valueReference.identifier.value                         AS provider_npi
FROM g;
""")

diff = con.execute("""
WITH api AS (
    SELECT p.mrn, p.last_name, p.dob, m.*
    FROM api_members m JOIN api_patients p ON p.id = m.patient_id
    WHERE NOT m.inactive
)
SELECT
    coalesce(c.mrn, a.mrn) AS mrn,
    CASE
        WHEN a.mrn IS NULL                    THEN 'missing_in_api'
        WHEN c.mrn IS NULL                    THEN 'missing_in_csv'
        WHEN c.provider_npi <> a.provider_npi THEN 'provider_mismatch'
        WHEN c.period_start <> a.period_start
          OR c.period_end   <> a.period_end   THEN 'period_mismatch'
        WHEN c.dob <> a.dob
          OR c.last_name <> a.last_name       THEN 'demographics_mismatch'
    END AS issue
FROM csv_members c
FULL OUTER JOIN api a USING (mrn)
WHERE issue IS NOT NULL
ORDER BY issue, mrn;
""").df()

print(diff.groupby("issue").size().to_string())
diff.to_csv("recon_2026-10-31.csv", index=False)

Why DuckDB and not a pandas merge: the FHIR side is nested (identifiers, extensions, periods), and list_filter + unnest flatten it in SQL you can read in a review. And it scales to a few million rows on a laptop.

Classifying differences

A raw diff count is useless; you need a category per row and a rule per category.

Track unexplained diffs, not total diffs. Expected ones (a known timing lag in phase 2) get an allow-list with an expiry date; everything else pages someone.

Exit criteria (write them down before you start)

• Zero unexplained diffs for N consecutive daily runs (I use 20 business days, covering a month-end).
• At least one full month-end close where finance and quality reporting ran off the API data and matched prior method.
• Notification channel healthy: no missed heartbeats over the window (Part 6).
• Rollback rehearsed: re-enable SFTP as source of truth from a phase-3 state, with the generated CSV, in under an hour.
• Every consumer has a named owner who confirmed in writing they no longer read the SFTP folder.

Decommissioning SFTP properly

• Set the inbound folder read-only first; watch for failed uploads for two weeks — those are the producers you missed.
• Archive the historical files per your retention policy (they’re PHI), then remove access.
• Revoke the SSH keys. All of them. This is the moment the static-credential risk from Part 1 actually ends.

Series recap

All of it on public standards (HL7 Da Vinci ATR, FHIR Bulk Data, SMART, Subscriptions backport), the same model CMS uses for ACO beneficiary lists in BCDA.

References

• Da Vinci ATR IG
• ATR use cases — BCDA and DPC
• DuckDB — reading JSON / NDJSON
• DuckDB — CSV import
• DuckDB — list functions (list_filter)
• DuckDB — unnest
• Martin Fowler — Parallel Change
• Martin Fowler — Strangler Fig Application
• FHIR Bulk Data Access IG

31 Oct 2026 · #fhir #dataengineering #duckdb #rosterapiseries

GLB_