Data Ingestion

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

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