# AGENTS.md: PySpark pipelines

## Commands
- Tests: `pytest -q` (uses a local `SparkSession` fixture with `master("local[2]")`)
- Lint/format: `ruff check . && ruff format --check .`
- Run a job locally: `spark-submit --master local[4] jobs/<job>.py --date 2026-01-01`
- Never point local runs at production tables; use the `dev` catalog/schema from config.

## Code map
`jobs/` entrypoints (thin: parse args, build session, call transforms) · `src/<pkg>/transforms/` pure functions `DataFrame -> DataFrame` · `src/<pkg>/io.py` reads/writes · `tests/` with small in-memory DataFrames.

## Rules
- Use the DataFrame API and `pyspark.sql.functions as F`. Avoid Python UDFs; if unavoidable, use vectorized `pandas_udf` with declared return types.
- Define schemas explicitly (`StructType`) when reading JSON/CSV; never rely on `inferSchema` in production.
- No `collect()`, `toPandas()` or `count()` on large data inside pipelines; they pull data to the driver or trigger extra jobs. Use them only on aggregated results or in tests.
- Joins: broadcast small dimension tables with `F.broadcast`; check for skew on big keys; always specify join type.
- Partition writes by low-cardinality date columns; avoid tiny-file explosions (`repartition`/`coalesce` before write, or optimize/compaction on the table format).
- Writes are idempotent: Delta/Iceberg `MERGE` or dynamic partition overwrite for a run date, never blind `append` of reprocessed data.
- Column names are `snake_case`; select columns explicitly instead of `select("*")` at stage boundaries.
- `cache()`/`persist()` only when a DataFrame is reused, and `unpersist()` after.
- Config (paths, catalogs, dates) comes from arguments or config files, not hard-coded strings.

## Tests
Use `pyspark.testing.assertDataFrameEqual` (or chispa) on tiny fixtures covering nulls, duplicates and late data.

## Done
Lint and tests pass, and the job runs locally on sample data end to end.