18 min read
By

I Tried to Make Structured File Intake Boring

Data engineering ETL Data Quality Idempotency Product architecture

I wanted one YAML file to be enough to onboard an ordinary file source.

That sentence is how ingestion platforms get out of hand. First it means CSV options and a target table. Then it means a schema registry, a connector service, a rule language, a scheduler, a state machine, and a UI for explaining the state machine. Six months later, the original CSV still needs custom code.

So I wrote the exclusions before I wrote the framework.

No scheduler. No watcher. No connector service. No separate metadata database. No plugin registry. No general replay system. I built it inside a notebook-driven data project that already had local files, a SQL database, job execution, retained output, and quality checks. File intake had to use those pieces rather than become a product around itself.

The actual claim was narrower:

A normal local file source should be onboarded mostly through configuration while getting stable identity, validation, duplicate protection, rejected-row handling, provenance, and safe retries.

I tried it against five source definitions from three domains that have very little reason to resemble one another: clinical-trial data, a U.S. provider registry, and public-transit schedules.

The normal path worked. The first full load wrote its data, then the surrounding job failed while retaining notebook output. By then the intake attempts were complete, so the rerun discovered all ten files and wrote no additional rows. I tested the harder boundary separately: a crash after landing rows were written but before the attempt was finalized.

Those two recovery paths changed what I trust about the design more than the happy path did.

Three domains that should have pulled the design apart

I kept the fixtures tiny and deterministic because this was an architecture probe, not a throughput benchmark.

Clinical trial files. Clinical studies often separate data by subject and by kind of observation. The synthetic fixture borrows that shape from CDISC’s Study Data Tabulation Model: a demographics-style file identifies the people in the study, a laboratory file records test results, and an adverse-events file records medical events reported for a subject during the trial, whether or not the treatment caused them. In this fixture, demographics and labs share one ZIP while adverse events arrive as a separate CSV. That made it useful for testing related datasets, required identifiers, invalid measurements, and corrections without pretending to implement the full clinical standard.

NPPES provider data. NPPES stands for the National Plan and Provider Enumeration System. It is the CMS system behind National Provider Identifier records for health-care providers in the United States. CMS publishes downloadable NPPES data as monthly full files and weekly incremental ZIPs. A package can include the main provider records plus reference files for other names, practice locations, and endpoints. That combination gave me a very different test: a potentially large central CSV, several package members, periodic increments, renamed duplicates, and a corrected weekly delivery.

GTFS transit feeds. GTFS stands for the General Transit Feed Specification, an open format transit agencies use to publish schedule data for journey planners and other rider-facing applications. A GTFS Schedule feed is usually one ZIP containing linked text tables: routes, trips, stops, stop times, service calendars, and optional supporting files. The filenames are conventional, and IDs connect one table to another. It is almost the ideal compact test for required package members and cross-file consistency: a trip should point to a route that exists, and a stop time should point to both a trip and a stop that exist.

The files describe completely different things, but the intake problems underneath them are surprisingly similar.

I did not build an SDTM engine, an NPPES downloader, or a complete GTFS validator. That would have tested three domain products. I wanted to know whether the same ordinary intake path could survive all three with configuration changes only.

Project files and one YAML source definition enter a fixed intake path that records delivery identity and attempts, stages and validates rows, sends accepted rows to landing tables with provenance, retains rejected rows, and records execution evidence.

Configuration describes the source. Identity, state, rejection, and retry behavior stay fixed.

What run_source() actually does

The public surface is one function, but the function is not a black box. It follows the same path for a clinical CSV, an NPPES ZIP, and a GTFS feed. Three nouns keep that path understandable:

  • a source is the configuration and inbox being processed;
  • a delivery is one unique set of bytes observed for that source; and
  • an attempt is one execution against a delivery.

That separation matters because the same delivery can be seen under another filename, attempted again after a crash, or replaced by corrected bytes. Here is the complete path through the framework.

  1. Load and normalize the source definition. run_source("gtfs"), for example, resolves config/intake/gtfs.yaml, validates the supported keys, fills the small set of defaults, and serializes the normalized result. Its SHA-256 becomes the configuration hash, while the complete normalized JSON is retained on every attempt. An old attempt can therefore be understood without a separate schema registry or configuration-snapshot service.

  2. Discover files when the job runs. The framework scans the configured local path and pattern. There is no watcher or background poller. If nothing matches, it records a NO_WORK attempt with no delivery and returns normally. No file is invented merely to give an empty run something to process.

  3. Stream the delivery identity. For every matching file, the framework reads the bytes in chunks and calculates SHA-256. The durable identity is source_id + content_sha256; the filename is only descriptive metadata. This is why a renamed copy of the same NPPES archive is a duplicate, while a new GTFS feed published again as transit.zip is new content.

  4. Register the delivery and classify what was found. New bytes create a PENDING delivery. If the same source and hash already exist in a completed state, the framework creates a DUPLICATE attempt and stops before parsing or writing rows. When a source has a logical delivery_key, different bytes for the same key are treated as a candidate correction. The new delivery points to the currently loaded one, but the old delivery is not marked SUPERSEDED until the replacement finishes successfully.

  5. Create the attempt before touching landing data. The attempt captures its execution ID, start time, delivery, configuration hash, and complete normalized configuration. If an older attempt for this delivery never reached a terminal status, the new run first closes it as FAILED with an INTERRUPTED code. That turns an abandoned write into visible history instead of an invisible half-run.

  6. Clear this delivery’s partial landing rows. Before loading a non-duplicate delivery, the framework deletes accepted rows carrying that delivery ID from each configured target. This is the retry boundary. If a process died after rows were persisted but before the attempt was finalized, the next attempt starts that delivery’s landing scope cleanly and writes one final copy.

  7. Open the package safely, then select dataset members. A plain file goes directly to its configured reader. A ZIP is extracted to a temporary directory after rejecting absolute paths, traversal paths, too many members, and excessive declared or extracted size. The original archive is never changed. Each dataset then matches its configured member name or pattern; an absent optional member is noted, while an absent required member invalidates the delivery.

  8. Read into deterministic staging. CSV, delimited text, JSON, NDJSON, and Parquet use a small reader dispatch and DuckDB relations. GTFS .txt files are simply delimited inputs. Each source record receives _intake_row_number, and typed columns use safe casts while the raw record remains available for rejection evidence.

  9. Turn validation into failing row numbers. A missing required column is a delivery-level structural failure: the framework records the blocking quality check, marks the attempt and delivery INVALID, and raises so the surrounding run fails clearly. Type failures and the five supported row rules—not_null, accepted_values, range, unique, and sql—produce sets of failing row numbers. Their union decides the split once; there is no expression language or rule-engine state machine hiding behind it.

  10. Persist both sides of the split. Rows without a failure go to the configured landing table with delivery, attempt, execution, ZIP-member, row-number, and load-time provenance. Failing rows go to intake.rejected_record with the original record, member, row number, rule IDs, and reason. Before an attempt can complete, the framework enforces rows_read = rows_accepted + rows_rejected.

  11. Finalize state only after the data path completes. A successful attempt becomes LOADED, as does its delivery. Only then is the older same-key delivery changed to SUPERSEDED. An unexpected exception finishes the attempt as FAILED and is re-raised. The caller receives delivery-level results and row totals, while the durable explanation remains in intake.delivery, intake.attempt, and intake.rejected_record.

That gives each common outcome one unambiguous stopping point:

  • No matching file: record a NO_WORK attempt and write no landing rows.
  • Same source and same bytes: record a DUPLICATE attempt and write no landing rows.
  • Missing required file or column: mark the attempt and delivery INVALID, fail the run, and remove any partial landing rows before the next attempt.
  • Tolerated type or rule failures: complete the attempt as LOADED, land the accepted rows, and retain the rejected records separately.
  • Same logical key with successfully loaded new bytes: mark the new delivery LOADED and the old delivery SUPERSEDED.
  • Unexpected execution failure: mark the attempt FAILED, fail the run, and remove and replace that delivery’s partial rows on retry.

The YAML controls what to discover, read, validate, and target. It does not control how identity, history, rejection, supersession, or retry safety work. Those are framework guarantees.

The YAML describes the file, not a new programming language

Each source keeps its discovery, package, datasets, columns, rules, targets, and correction behavior together. These are the representative definitions from the experiment. The clinical tab shows its multi-dataset lab package. The NPPES tab includes both the monthly full source and the weekly incremental source so the correction key is visible.

version: 1
id: clinical_lab
name: Synthetic CDISC-style laboratory delivery

discovery:
  path: files/inbox/clinical
  pattern: "lab_*.zip"

delivery_key:
  filename_regex: "lab_(?P<date>[0-9]{8})"

package:
  type: zip
  required_members: [dm.csv, lb.csv]

datasets:
  - id: subjects
    file: dm.csv
    reader: {type: csv}
    target: {schema: raw, table: clinical_subjects}
    columns:
      study_id: {type: varchar, required: true}
      subject_id: {type: varchar, required: true}
      site_id: {type: integer}
    rules:
      - {id: subject_required, type: not_null, column: subject_id, action: reject_row}
      - {id: subject_unique, type: unique, column: subject_id, action: reject_row}

  - id: lab_results
    file: lb.csv
    reader: {type: csv}
    target: {schema: raw, table: clinical_lab_results}
    columns:
      study_id: {type: varchar, required: true}
      subject_id: {type: varchar, required: true}
      test_code: {type: varchar, required: true}
      result_value: {type: double}
      collected_at: {type: timestamp}
    rules:
      - {id: subject_required, type: not_null, column: subject_id, action: reject_row}
      - id: valid_test
        type: accepted_values
        column: test_code
        values: [GLUC, HBA1C, CREAT]
        action: reject_row

behavior:
  same_key_new_content: supersede

The GTFS configuration points the delimited reader at .txt members and adds one SQL rule for cross-file consistency. The NPPES monthly source gets byte-level duplicate protection; the weekly source adds a date-based delivery key so new bytes for the same week become a correction. The clinical package maps two related CSVs. In every case, raw is simply the landing schema. The framework code does not change.

I stopped the rule vocabulary at not_null, accepted_values, range, unique, and sql. The SQL option is the escape hatch. It kept me from inventing a small, worse version of SQL and calling it a rule DSL.

The example pipeline really does call the shared framework

I kept the reusable code and the proof in separate projects. The framework project owns intake.py; the examples project owns the five YAML definitions, fixture files, and one ingestion notebook. That ingestion notebook is deliberately boring glue: it runs the framework notebook once, takes the returned run_source function, and calls it for each configured source.

The Intake Framework project exposes run_source from intake.py. The Intake Examples job loads that notebook once, calls run_source for five YAML source IDs, and writes accepted rows plus intake state to its own DuckLake catalog.

The examples project depends on the framework project. It does not contain a second ingestion implementation.

This is the important part of ingest_sources.py:

from anatini_projects.intake_framework.notebooks.intake import (
    app as intake_framework,
)


_, framework = intake_framework.run()
run_source = framework["run_source"]


source_ids = [
    "clinical_lab",
    "clinical_adverse_events",
    "nppes_monthly",
    "nppes_weekly",
    "gtfs",
]


results = [run_source(source_id) for source_id in source_ids]

For each ID, run_source() loads config/intake/<source_id>.yaml, scans that source’s inbox, and applies the fixed identity, validation, rejection, provenance, and retry path. Accepted rows and Intake state stay in the examples project’s own DuckLake catalog. The separate portfolio_results.py notebook only queries that retained state; it does not perform ingestion.

There is no registry class hierarchy behind the public function. Reader selection is an unsurprising if branch for CSV, delimited text, JSON, NDJSON, and Parquet. ZIP is a package wrapper, not another reader abstraction.

The things I refused to make configurable

The experiment became simpler once I separated source description from system guarantees.

Identity is the delivery bytes

Every delivery is identified by:

source_id + SHA-256(delivery bytes)

The hash is streamed so a large file does not have to sit in Python memory. The filename remains metadata.

This follows the same rule I wrote about in Checksum, Not Filename. An identical weekly NPPES ZIP under a different name is still a duplicate. A GTFS package named transit.zip with different bytes is not.

The optional delivery_key answers a different question. It can come from the filename or a filename regex and means “these deliveries occupy the same logical slot.” Same key and same hash is a duplicate. Same key and a different hash is a correction, so the new delivery links to the old one and the old one becomes SUPERSEDED.

I did not make hashing pluggable. None of the three examples needed another identity provider.

A delivery is not an attempt

The durable model has three tables:

TableWhat it remembers
intake.deliverysource, content identity, logical key, current state, and supersession link
intake.attemptexecution ID, timestamps, status, row counts, normalized configuration, summary, and error
intake.rejected_recorddataset, ZIP member, row number, failed rules, reason, and original record

That distinction matters. A delivery can be observed once and attempted more than once. A duplicate is still an attempt worth recording. A crash can leave an incomplete attempt without changing what the delivery bytes are.

I nearly added a state-event table. I left it out because the current delivery state, attempt history, and supersession link answered every question the fixtures asked. “Maybe useful later” was not enough.

Every row gets one deterministic address

Each dataset first becomes a staging relation with _intake_row_number. Try-casts and rules return failing row numbers. The union of those failures splits the staging relation once:

rows read = rows accepted + rows rejected

Accepted rows receive six provenance columns: delivery ID, attempt ID, execution ID, ZIP member, source row number, and load timestamp. Rejected rows go to the one shared reject table. There are no per-source reject schemas to hunt through.

A missing required column is different from a bad value in one row. The missing column invalidates the delivery and fails the execution. A value that cannot be cast, or a tolerated rule violation, records a warning-level quality check and rejects only that row.

That is the behavior I want when a file is structurally wrong: a loud failure, not the quiet column shift I described in The Missing Column That Didn’t Crash.

ZIP support was where “small” stopped meaning casual

NPPES and GTFS both needed ZIP packages, so ZIP earned a place in the baseline. General archives did not.

The package handler rejects absolute member paths and path traversal, limits member count, limits total uncompressed size, extracts into a temporary directory, and never modifies the original archive. Tests cover each of those limits plus a missing required member.

The framework records the member path on accepted and rejected rows and in the attempt summary. I considered a durable delivery_file table, but the three examples did not need one. The delivery remains the top-level object.

This is a small distinction with a useful consequence: package safety is centralized, but the framework does not pretend ZIP, tar, 7z, nested archives, and remote object listings are one clean abstraction.

Two failures, two retry boundaries

The first full fixture execution read 40 rows. It accepted 38 and retained 2 rejected lab records. It also recorded 55 quality checks: 53 passed, while 2 warning-level checks identified the rejected values.

Then the surrounding job failed after ingestion, while retaining the notebook output on Windows.

That was inconvenient, but it happened outside the intake path. The delivery attempts had already been finalized successfully.

On the next execution, the saved job rediscovered all 10 physical files and accepted 0 new rows. Two of those files were renamed byte-identical copies, so the 10 paths resolved to 8 content identities: 5 current and 3 superseded. Repeated verification runs eventually accumulated 52 DUPLICATE attempts while the loaded row counts stayed at 40 read, 38 accepted, and 2 rejected.

Retained evidenceCount
configured sources with deliveries5
physical files discovered per full scan10
content-identified deliveries8
current loaded deliveries5
superseded deliveries3
successful-load rows read40
accepted rows38
rejected rows2
duplicate attempts after verification reruns52

The incomplete-attempt case was a separate deterministic test. It deliberately crashes after landing rows are written but before the attempt is finalized. The rerun recognizes the incomplete attempt, clears rows for that delivery scope, and writes one final copy. The focused framework suite has 19 passing tests, including the crash, renamed duplicates, supersession, invalid schemas, row rejection, no-work runs, unsafe ZIPs, and all five example configurations.

This is safe retry, not historical replay. I did not add a button that deletes a successful old load and tries again. As How Twelve Rejected Files Became Unreloadable argues, requeue, replay, and backout are different destructive operations. This baseline was not the place to blur them together.

What stayed out

The framework scans when a user runs the notebook or its saved job executes. It does not watch directories or poll upstream systems. Acquisition is a separate streaming HTTP helper that places a file locally and records URL, retrieval time, hash, and size.

There is also no:

  • workflow engine or scheduler inside the framework;
  • source-specific UI;
  • configuration snapshot table—the normalized configuration lives on the attempt;
  • schema evolution machinery;
  • archive plugin system;
  • reader registration framework; or
  • notification service.

Those omissions are not claims that the features are useless. They are the answer to one question I kept asking during the build:

Do at least two of the three real examples require this now?

If the answer was no, I tried to leave a clean function boundary and move on.

Where this answer stops travelling

This is a proof of architecture using tiny synthetic fixtures. It does not establish NPPES throughput, memory use on a national provider archive, or correctness against the complete GTFS specification. I have not run the optional official large NPPES exercise yet.

The byte hash is deliberately literal. Recompress the same ZIP members with different archive metadata and the package bytes change, so Intake sees new content. If an upstream system repackages identical logical data constantly, exact delivery identity may be too sensitive. That would need evidence and a source-specific normalization decision; I would not quietly redefine identity for every source.

The design also assumes bounded local files and one project-owned analytical database. I would choose a different boundary for streaming events, database change capture, nested archives, remote object-store discovery, several concurrent writers, or a source whose parsing genuinely requires custom code.

And while the examples cover safe retry, they do not provide a general way to replay successful historical deliveries. That omission becomes unacceptable the moment operators need routine backouts or corrected transformation logic. At that point I would design the destructive workflow explicitly, with previews and a named blast radius, rather than smuggling it into run_source().

The part I would reuse

The reusable idea is not the YAML shape. It is deciding which choices a source owns and which guarantees the intake path owns.

The source owns where its files are, what members and columns it expects, which small set of rules applies, and where accepted rows land. The intake path owns byte identity, delivery and attempt history, deterministic row addressing, provenance, rejection, and retry convergence.

That division was enough for clinical, NPPES, and GTFS-shaped data without turning their differences into framework code. The acceptance test for a sixth ordinary CSV source is simple: one YAML file plus the file itself, with no framework edit. An unusual source can still use Python.

That is about as universal as I want this framework to become.

Follow the notebook evidence

The static marimo snapshots follow the same path as the experiment. Each opens in a new tab, so the article stays beside the retained output:

  1. Framework self-test confirms that configuration loading, streaming identity, and source execution are available through the small public surface.
  2. Ingestion Job and duplicate-safe rerun shows the examples pipeline loading the framework, calling run_source() for all five source definitions, and refusing to load accepted rows twice.
  3. Verified results reads the retained state and presents delivery status, attempt outcomes, supersessions, rejected rows, and accepted-row provenance.

These are rendered notebook outputs, not screenshots. They are static, contain no notebook source, and do not execute the ingestion again. The same HTML files are also available in the public output bundle for download. The fixtures, workspace database, internal execution bundles, and notebook source are intentionally not published.

More on Data engineering Checksum, Not Filename →