data-ingestion

name

data-ingestion

description

Load whatever the request points at — one CSV or Excel export, twelve related tables, a folder of 500 emails, a 4 GB log you never load, a warehouse table you query in place — and record every source in run/manifest.json with the grain one row represents, the axis it is ordered by, its row count, columns and dtypes. Read as text and coerce on purpose — pd.read_csv inference turns order_id 00123 into the integer 123, and a paged pull that stops early looks identical to a complete one. Use when a user attaches a file or a folder, points you at a database, warehouse or API, or asks for analysis of data you have not loaded yet.

Data ingestion

A load that "worked" is the most expensive failure in an analysis. Nothing raises, a frame comes back, and four stages later a number is wrong because order_id lost its leading zeros, a Total row became a data row, or the API returned page one of forty. You decide the shape everything downstream assumes; run/manifest.json is where you say so out loud.

Applies to whatever the request points at: one flat export, twelve related tables, a folder of 500 emails, a 4 GB log you never load, a warehouse table you query in place. One entry per source — every table, directory of parts and warehouse query gets its own key in run/manifest.json, with a recorded query and a verified count where the bytes are not yours to keep. Enter here when data is not loaded and recorded yet; stop here when "what did I just get" was the question; already-typed data skips the text read, not the entry.

Read as text, then coerce on purpose

Type inference is where downstream bugs are born. pd.read_csv with no dtype= reads 00123 as the integer 123, deleting the zeros that make it an identifier, and turns a column holding one N/A into float64. Read as string, coerce only what you compute on, and count the casualties:

import pandas as pd

text = pd.read_csv("run/raw/orders.csv", dtype="str",
                   keep_default_na=False, na_values=[""])
df = text.copy()
df["qty"] = pd.to_numeric(text["qty"], errors="coerce").astype("Int64")
df["created_at"] = pd.to_datetime(text["created_at"], format="ISO8601",
                                  errors="coerce", utc=True)
for col in ("qty", "created_at"):
    lost = text[col].notna() & df[col].isna()
    if lost.any():
        print(col, int(lost.sum()), "unparseable:",
              text.loc[lost, col].head(3).tolist())

keep_default_na=False keeps NA, None and null as literal text, not someone else's idea of missing; errors="coerce" makes a bad value NaT/<NA> and the loop stops it being silent. An ID you can do arithmetic on is a bug.

Size and sniff the source before you load it

ls -l run/raw/orders.csv && head -c 400 run/raw/orders.csv   # .gz: zcat f.gz | head -c 400

That shows the real delimiter and the header depth; ls -R and unzip -l do the same for a folder or an archive, and file-handling owns never reading a whole data file to find out. Past a few GB nothing loads, and decompressing to make it load fills the disk: leave the .gz alone, count from it with duckdb.sql("SELECT count(*) FROM 'log.csv.gz'").df(), and fold a chunked pd.read_csv in fixed memory.

Every source has one trap, and this is it

Source	How to read it	The trap
CSV / TSV, or a table pasted into the message	pd.read_csv(..., dtype="str"); a spreadsheet paste is tab-separated, so sep="\t"	European exports: sep=";", decimal=",", thousands="." — the default read gives one column, or numbers as text; a survey's second header row of question wording, or a pasted Total row, arrives as data — peek first, as with Excel
Excel .xlsx	pd.ExcelFile, then pd.read_excel(header=n)	it is a formatted report, not a table — title rows, spacer rows, a Total row that becomes data
JSON / JSONL	one document is json.load(f) then pd.json_normalize(doc["features"], sep="__"); JSONL that fits in memory is json.loads per line into pd.json_normalize(recs, sep="__"), and past that project only the columns you need with duckdb.read_json(path, columns={...}) — a .gz reads in place	pd.read_json(lines=True) "succeeds" and hands you a column of dicts; nested lists survive normalization as Python lists, so hold a geometry or a tag list out of the frame; a list comprehension of json.loads over a 4 GB log OOMs before it raises, and an auto-inferred schema silently drops keys absent from its sample
SQL file: .sqlite, or a .sql dump	sqlite3 or duckdb, then query it	SELECT * on a view, or on a table with soft-deleted rows still in it; a dump is DDL + INSERTs in a vendor dialect, and regexing INSERT lines drops any row with an escaped quote
Warehouse or live database	query in place: pd.read_sql on a bounded SELECT, count(*) for the row count; only duckdb and pyarrow ship here, so pip install sqlalchemy plus the vendor driver (psycopg2-binary, snowflake-sqlalchemy, sqlalchemy-bigquery — a bare postgresql:// DSN resolves to psycopg2, not psycopg 3) and take the DSN from an agent env credential, never from chat — workspace-setup owns adding one	extracting the table to get a frame — you cannot, and the manifest wants the query and the DSN with its credentials stripped, not a sha256
Paged API	the loop below	a short final page and an early stop look exactly like completion
Many parts: a folder, a zip, 12 sheets in one workbook, a 12-table export	same-shaped parts stack into one entry — pd.read_excel(xl, sheet_name=None), then pd.concat(parts, names=["plate"]).reset_index(level="plate"), because the part's name is data; compare the parts' columns and dtypes before you stack, and differently-shaped tables get one entry each with the join left to data-cleaning	the part name discarded, so no row knows which sheet it came from; part 007 read with different dtypes than its siblings; or a join done in whatever cell needed it and logged nowhere
A collection: 500 emails, images, JSON documents	leave the items where they are and build an index frame — one row per item, carrying path, an item id and the attributes you pulled out; a collection can be one document rather than many files, and then a GeoJSON feature's own id stands in for the path	forcing the corpus itself into a frame. Decoding the domain — image internals, audio, protein sequences — is the analysis's job, not ingestion's

An Excel sheet is a report, not a table — peek before you choose header=:

import pandas as pd

xl = pd.ExcelFile("run/raw/report.xlsx")   # xl.sheet_names lists every tab
print(pd.read_excel(xl, sheet_name="Q3 Summary", header=None, nrows=8).to_string())
df = pd.read_excel(xl, sheet_name="Q3 Summary", header=3, dtype="str")
df = df[df["Region"].notna() & (df["Region"] != "Total")]

A paged pull is incomplete until you have proved it complete

The worst outcome here is a partial pull that looks whole — half the rows analyze cleanly and every conclusion is wrong. Assert the count you got against the source's own total, and never read a rate limit as the end of the data:

import os, time, requests

url = "https://api.example.com/v1/orders?page=1&per_page=100"
headers = {"Authorization": f"Bearer {os.environ['API_TOKEN']}"}
rows, body = [], None
while url:
    r = requests.get(url, headers=headers, timeout=30)
    if r.status_code == 429:
        time.sleep(int(r.headers.get("Retry-After", "5")))
        continue
    r.raise_for_status()
    body = r.json()
    rows.extend(body["data"])
    url = body.get("next_page_url")
assert len(rows) == body["meta"]["total"], f"partial pull: {len(rows)} rows"

If no total is reported, say completeness is unverified. A 404 stops the run, and a missing DSN or token is requested through workspace-setup rather than pasted into chat — never invent a plausible frame.

What you write: run/raw/ and run/manifest.json

A chat attachment is a signed URL: curl -sS "<url>" -o run/raw/<name>, so run/raw/ holds the bytes as retrieved — there is none for a warehouse table you query in place. Later stages write beside it; none edit what you wrote.

import datetime, hashlib, json, pathlib
src = pathlib.Path("run/raw/orders.csv")
pathlib.Path("run/manifest.json").write_text(json.dumps({"orders": {
    "source": "https://api.example.com/v1/orders",
    "retrieved_at": datetime.datetime.now(datetime.UTC).isoformat(),
    "grain": "one row per order line, ordered by created_at",
    "rows": len(df),           # or a duckdb count(*), when nothing was loaded
    "cols": list(df.columns),  # just the projection, when that is all you read
    "dtypes_as_read": {c: str(t) for c, t in df.dtypes.items()},
    "sha256": hashlib.file_digest(src.open("rb"), "sha256").hexdigest(),
}}, indent=2))

Required per entry: source, retrieved_at, grain, rows, cols, dtypes_as_read, and either a sha256 — for a directory of items, the item count and the decode failures you verified instead — or, bytes not yours, the query plus a count checked against the source's own total. grain is the sentence "one row of this represents ___, ordered by ___": per (store, week), by week. Ask when the source will not say. Name the ordering column too: validation and EDA bucket along the axis you name, and a composite grain has more than one.

Hand the manifest to data-validation; you asserted only that the data is loaded and counted, so if the answer ships from here, end the reply with the literal Stages: … · skipped: … line eda-storytelling defines. A 40-row paste needs no run/; the manifest is one sentence in your reply.

Common failure modes

Symptom	Likely cause	Fix
UnicodeDecodeError, or names arrive as é / id	the export is cp1252, or has a UTF-8 BOM	encoding="cp1252" or encoding="utf-8-sig", chosen from the peek
df["Region"] raises KeyError though the header looks right	trailing whitespace in the header cells	df.columns = df.columns.str.strip() before anything else
Dates all land in the first 12 days of each month	d/m vs m/d resolved per value	pass an explicit format=, and count the NaTs the coerce produced
A date column holds 45231	Excel serial dates	pd.to_datetime(s.astype("Int64"), unit="D", origin="1899-12-30")
df.dtypes == object selects no columns	pandas 3 reads text as str dtype, not object	pd.api.types.is_string_dtype(s), or df.select_dtypes("str")
Two pulls minutes apart return different row counts	a live source paginating over a moving table	sort by a stable key server-side; retrieved_at in the manifest is what dates the answer