Home › Apache Spark / PySpark › PySpark ETL AGENTS.md

PySpark ETL AGENTS.md

AGENTS.md for PySpark data pipelines: DataFrame API over UDFs, explicit schemas, partitioning, Delta/Iceberg writes, and local pytest with SparkSession.

AGENTS.mdApache Spark / PySpark

Copy AGENTS.md

How to use

Save this as AGENTS.md in your repository root, commit it, and your agent will pick it up automatically. Edit the commands to match your project.

Note: Review and test before use. Provided as is, without warranty. See Terms.

# 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.