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: ```python 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 ```bash 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=`: ```python 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: ```python 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. ```python 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 `NaT`s 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 |