DBT: S3 File Schemas
DBT model name: s3_file_schemas
Explore dependencies/lineage: link
Description
Detected column schema per (key, etag) for .parquet (native types) and .csv / .xlsx (read with all_varchar + a tiny sample for robust column discovery; CSV encoding tried UTF-8 then ISO-8859-1; legacy .xls not supported by DuckDB). Re-sniffed only on ETag change.
Details
| Column | Type | Description |
|---|---|---|
columns_json |
`` | JSON array of {name, type} detected for the file. |
sniff_error |
`` | Populated when detection failed (non-fatal; file is still listed). |
No data tests defined 🧐
Not referenced by any model or exposure.
"""Detected column schemas for parquet / CSV / Excel files in `trase-storage`.
Standalone, reusable table: one row per (key, etag). The goal is robust **column
discovery**, not precise typing — and to do it **without downloading whole
files** (some objects are gigabytes):
* parquet → native types via a clean DuckDB + httpfs connection, which reads
only the parquet **footer** over HTTP range requests (pyarrow
footer-read fallback for files DuckDB rejects, e.g. GeoParquet).
* csv → only the first `s3_monitor_schema_head_bytes` are fetched with a
boto3 ranged GET, line endings normalised, then DuckDB auto-detect
(`all_varchar`, UTF-8 then ISO-8859-1). If DuckDB errors or
collapses to one column (an unquoted delimiter inside a field),
fall back to splitting the header on the most frequent delimiter.
* excel → .xlsx can't be partially read (zip), so the whole object is
fetched only when `size_mb <= s3_monitor_schema_max_excel_mb`,
else recorded as a (non-fatal) sniff_error. Legacy .xls is not
supported by DuckDB's read_xlsx, so it is not sniffed.
We deliberately do NOT sniff through the dbt `session`: the production profile
registers a cached fsspec S3 filesystem that pulls entire objects. A dedicated
`duckdb.connect()` here uses native httpfs range reads instead.
Incremental and keyed on (key, etag): a file is only (re-)sniffed when new or its
ETag changed. Each run is capped at `s3_monitor_schema_sniff_batch` newly-seen
files (newest first). Any read/detection error is recorded in `sniff_error` so a
single unreadable file never fails the model.
"""
import json
import os
import tempfile
import duckdb
import pandas as pd
from trase.config import aws_session
from trase.tools.progress import ProgressPrinter
# Only modern OOXML .xlsx — DuckDB's read_xlsx cannot read legacy BIFF .xls.
_CSV_EXTS = ("csv",)
_EXCEL_EXTS = ("xlsx",)
OUT_COLS = [
"key",
"etag",
"file_type",
"encoding",
"delimiter",
"quote",
"escape",
"has_header",
"skip_rows",
"n_columns",
"columns_json",
"sniffed_at",
"sniff_error",
]
def _blank_row(key, etag, file_type):
return {col: None for col in OUT_COLS} | {
"key": key,
"etag": etag,
"file_type": file_type,
}
def _columns_from_describe(df) -> tuple[int, str]:
cols = [
{"name": n, "type": t} for n, t in zip(df["column_name"], df["column_type"])
]
return len(cols), json.dumps(cols)
def _q(path: str) -> str:
"""Single-quote-escape a path for inlining into SQL."""
return path.replace("'", "''")
def _sniff_parquet(rcon, s3_uri, row):
"""Footer-only schema read over native httpfs, with a pyarrow fallback for
files DuckDB rejects (e.g. GeoParquet missing a version in its metadata)."""
try:
df = rcon.execute(f"DESCRIBE SELECT * FROM read_parquet('{_q(s3_uri)}')").df()
row["n_columns"], row["columns_json"] = _columns_from_describe(df)
return
except Exception as exc: # noqa: BLE001
duck_err = str(exc)
try:
import pyarrow.fs as pafs
import pyarrow.parquet as pq
bucket, _, keypath = s3_uri[len("s3://") :].partition("/")
fs = pafs.S3FileSystem(region="eu-west-1")
schema = pq.read_schema(fs.open_input_file(f"{bucket}/{keypath}"))
cols = [{"name": n, "type": str(t)} for n, t in zip(schema.names, schema.types)]
row["n_columns"] = len(cols)
row["columns_json"] = json.dumps(cols)
except Exception: # noqa: BLE001
row["sniff_error"] = duck_err
def _guess_header_columns(body: bytes, encoding):
"""Fallback column discovery: split the first line on whichever of
`;` `\\t` `|` `,` occurs most often. Robust for files where DuckDB's
consistency-based sniffer fails — e.g. an unquoted delimiter inside a field
(common in free-text BOL descriptions), which makes DuckDB collapse to a
single column. Returns (delimiter, [column names])."""
first = body.split(b"\n", 1)[0]
text = None
for enc in [e for e in (encoding, "utf-8", "iso_8859_1") if e]:
try:
text = first.decode(enc)
break
except Exception: # noqa: BLE001
continue
if text is None:
return None, []
text = text.strip("\r")
count, delim = max((text.count(d), d) for d in (";", "\t", "|", ","))
if count == 0:
return delim, ([text.strip()] if text.strip() else [])
return delim, [c.strip() for c in text.split(delim)]
def _sniff_csv(rcon, s3, bucket, key, head_bytes, row):
"""Fetch only the first `head_bytes` via a ranged GET, normalise line
endings, and discover columns. DuckDB auto-detect is tried first (UTF-8 then
ISO-8859-1); if it errors or collapses to a single column, fall back to a
header-split heuristic."""
try:
body = s3.get_object(Bucket=bucket, Key=key, Range=f"bytes=0-{head_bytes - 1}")[
"Body"
].read()
except Exception as exc: # noqa: BLE001
row["sniff_error"] = f"range GET: {exc}"
return
# Normalise mixed CRLF/LF (which derails DuckDB's newline/dialect sniffer)
# and drop a partial trailing line.
body = body.replace(b"\r\n", b"\n").replace(b"\r", b"\n")
nl = body.rfind(b"\n")
if nl > 0:
body = body[:nl]
tmp = tempfile.NamedTemporaryFile(suffix=".csv", delete=False)
try:
tmp.write(body)
tmp.close()
local = _q(tmp.name)
duck_cols = None
enc_used = None
last_err = None
for enc in ("utf-8", "iso_8859_1"):
try:
df = rcon.execute(
f"DESCRIBE SELECT * FROM read_csv('{local}', "
f"all_varchar=true, sample_size=20, encoding='{enc}')"
).df()
duck_cols = list(zip(df["column_name"], df["column_type"]))
enc_used = enc
break
except Exception as exc: # noqa: BLE001
last_err = f"read_csv[{enc}]: {exc}"
head_delim, head_cols = _guess_header_columns(body, enc_used)
if duck_cols is not None and len(duck_cols) >= len(head_cols):
# Trust DuckDB (correct dialect, incl. quoting/escaping).
row["encoding"] = enc_used
row["n_columns"] = len(duck_cols)
row["columns_json"] = json.dumps(
[{"name": n, "type": t} for n, t in duck_cols]
)
try:
d = rcon.execute(f"FROM sniff_csv('{local}')").df().iloc[0]
row["delimiter"] = d.get("Delimiter")
row["quote"] = d.get("Quote")
row["escape"] = d.get("Escape")
row["has_header"] = (
bool(d["HasHeader"]) if d.get("HasHeader") is not None else None
)
row["skip_rows"] = (
int(d["SkipRows"]) if d.get("SkipRows") is not None else None
)
except Exception: # noqa: BLE001
pass
elif head_cols:
# Header-split fallback (DuckDB errored or collapsed to one column).
row["encoding"] = enc_used or "utf-8"
row["delimiter"] = head_delim
row["n_columns"] = len(head_cols)
row["columns_json"] = json.dumps(
[{"name": c, "type": "VARCHAR"} for c in head_cols]
)
if len(head_cols) <= 1:
row["sniff_error"] = last_err or "no delimiter detected"
else:
row["sniff_error"] = last_err or "could not read CSV header"
finally:
os.unlink(tmp.name)
def _sniff_excel(rcon, s3, bucket, key, ext, size_mb, max_excel_mb, row):
"""Excel is a zip, so it can't be partially read: download whole (only if
within the size cap) and read the schema locally with all_varchar."""
if size_mb is not None and size_mb > max_excel_mb:
row["sniff_error"] = f"skipped: {size_mb} MB > {max_excel_mb} MB excel cap"
return
try:
body = s3.get_object(Bucket=bucket, Key=key)["Body"].read()
except Exception as exc: # noqa: BLE001
row["sniff_error"] = f"GET: {exc}"
return
tmp = tempfile.NamedTemporaryFile(suffix=f".{ext}", delete=False)
try:
tmp.write(body)
tmp.close()
df = rcon.execute(
f"DESCRIBE SELECT * FROM read_xlsx('{_q(tmp.name)}', all_varchar=true)"
).df()
row["n_columns"], row["columns_json"] = _columns_from_describe(df)
except Exception as exc: # noqa: BLE001
row["sniff_error"] = str(exc)
finally:
os.unlink(tmp.name)
def model(dbt, session):
dbt.config(
materialized="incremental",
unique_key=["key", "etag"],
tags=["s3_inventory"],
)
# NOTE: tunables live under the model's +meta in dbt_project.yml. `meta` is a
# builtin dbt config, so dbt always propagates it from project config to the
# Python model (no need to statically reference each key, unlike custom config
# keys). Read the meta dict once, then index it. See README "Tunables".
meta = dbt.config.get("meta", {})
batch = int(meta.get("schema_sniff_batch", 2000))
head_bytes = int(meta.get("schema_head_bytes", 262144))
max_excel_mb = float(meta.get("schema_max_excel_mb", 50))
candidates = (
dbt.ref("s3_files_latest")
.filter("extension IN ('parquet', 'csv', 'xlsx')")
.project("bucket, key, etag, extension, size_mb, last_modified")
.df()
)
if dbt.is_incremental and not candidates.empty:
seen = session.execute(f"SELECT key, etag FROM {dbt.this}").df()
seen_pairs = set(zip(seen["key"], seen["etag"]))
mask = [
(k, e) not in seen_pairs
for k, e in zip(candidates["key"], candidates["etag"])
]
candidates = candidates[mask]
candidates = candidates.sort_values("last_modified", ascending=False).head(batch)
if candidates.empty:
return pd.DataFrame(columns=OUT_COLS)
# Dedicated read connection: native httpfs range reads (no cached fsspec).
rcon = duckdb.connect()
rcon.execute("INSTALL httpfs; LOAD httpfs;")
rcon.execute(
"CREATE SECRET (TYPE s3, PROVIDER credential_chain, REGION 'eu-west-1');"
)
try:
rcon.execute("INSTALL excel; LOAD excel;")
except Exception: # noqa: BLE001
pass
s3 = aws_session().client("s3")
rows = []
interval = float(meta.get("progress_interval_seconds", 60))
progress = ProgressPrinter(
len(candidates), "s3_file_schemas: sniffing", interval=interval
)
for rec in candidates.itertuples(index=False):
ext = rec.extension
size_mb = None if pd.isna(rec.size_mb) else float(rec.size_mb)
if ext in _EXCEL_EXTS:
file_type = "excel"
elif ext in _CSV_EXTS:
file_type = "csv"
else:
file_type = "parquet"
row = _blank_row(rec.key, rec.etag, file_type)
try:
if file_type == "csv":
_sniff_csv(rcon, s3, rec.bucket, rec.key, head_bytes, row)
elif file_type == "excel":
_sniff_excel(
rcon, s3, rec.bucket, rec.key, ext, size_mb, max_excel_mb, row
)
else:
_sniff_parquet(rcon, f"s3://{rec.bucket}/{rec.key}", row)
except Exception as exc: # noqa: BLE001 — belt-and-braces; never fail the run
row["sniff_error"] = str(exc)
rows.append(row)
progress.update()
progress.done()
rcon.close()
schemas = pd.DataFrame(rows, columns=OUT_COLS)
schemas["sniffed_at"] = pd.Timestamp.utcnow()
schemas["skip_rows"] = pd.to_numeric(schemas["skip_rows"], errors="coerce").astype(
"Int64"
)
schemas["n_columns"] = pd.to_numeric(schemas["n_columns"], errors="coerce").astype(
"Int64"
)
return schemas
-
Dbt path:
trase_ducklake.main_s3_monitoring.s3_file_schemas -
Containing yaml link: trase/data_pipeline/models/s3_monitoring/_schema_s3_monitoring.yml
-
Model file: trase/data_pipeline/models/s3_monitoring/s3_file_schemas.py
-
Tags:
s3_inventory