PySpark ETL Best Practices

.cursorrules for PySpark ETL

# PySpark ETL Project Rules

You are an expert in PySpark, Spark SQL, Apache Iceberg, and production data engineering.

## 1. Project Structure

- Build a `BaseETL` abstract class that owns the SparkSession lifecycle.
  Accept an optional `spark_session` for test injection.
- Keep `run_job()` as orchestration; each step is a named method.
  Compose steps with `.transform()`.
- Put CLI / argparse parsing in a standalone factory function. The
  config dataclass stays pure data so tests don't touch `sys.argv`.
- Build one shared `PartitionedReader` for date/hour/last-partition reads;
  keep business filters in the ETL where they are visible.

## 2. Code Style

- Use `import pyspark.sql.functions as F` and write `F.col()`, `F.when()`,
  `F.lit()` everywhere. Never use `df.col` attribute access — it breaks
  after joins and reassignments.
- Extract complex `.filter()` / `F.when()` conditions into named boolean
  variables once you exceed 3 expressions per clause.
- Prefer `select(...)` over `withColumn(...)` chains. `select` defines
  the output schema in one pass and skips the intermediate projections.
- Use `.alias()` instead of `.withColumnRenamed()` for renames.
- Max 5 statements per chain. Separate select/filter from
  withColumn and from join.

## 3. Joins

- Always specify `how=` explicitly. `.join(other, 'key')` is a bug.
- Prefer left joins over right joins. Flip the DataFrame order instead
  of writing a right join.
- Use `.alias('f')` / `.alias('p')` after a join and select via
  `F.col('f.col')`. Don't pre-rename every column.
- Wrap small dimension tables in `F.broadcast()` when Spark can't
  infer the size (post-filter tables, transformed dimension lookups).
- Never use `.dropDuplicates()` to mask a root cause. Find the duplicate
  source instead.

## 4. Window Functions

- Import `from pyspark.sql import Window as W`. Use `F` for functions.
- Always specify an explicit frame. Without one, Spark picks a frame
  that depends on whether `orderBy` is present — that's surprising.
- Know the difference: `row_number() + filter` drops rows,
  `first().over(w)` overwrites a column. Don't confuse them.
- Use `ignorenulls=True` with `first()` / `last()`. Without it, a null
  in the first row poisons the whole partition.
- Avoid empty `partitionBy()`. It forces a single partition.
  Use `.agg()` for global aggregations.

## 5. Map & Array Higher-Order Functions

- Use `map_zip_with` for merges with per-key conflict logic
  (timestamp-aware merges, "keep newer"). `map_concat` is fine only
  when there is no key overlap.
- Use `transform` + `array_max` to pull fields from nested structs
  (e.g. `array_max(transform(map_values(my_map), lambda x: x.ts))`).
- Avoid UDFs. Check for a built-in Spark function or higher-order
  function first — UDFs break Catalyst and add serialization cost.

## 6. Cumulative / Snapshot Tables

- Merges must be idempotent (same input, same output, no duplicates).
- Merges must be order-independent. Backfilling old data must not
  overwrite newer data. Use an explicit ordering criterion — not
  `coalesce` argument order.
- Validate primary key uniqueness and null-key checks after writes.

## 7. Data Quality & Performance

- Use `F.lit(None)` for empty columns. Never empty strings or "NA"
  sentinels.
- Don't use `.otherwise()` as a catch-all bucket. Unmapped values
  stay null so data-quality gaps surface.
- No `.show()`, `.collect()`, `.printSchema()` in production code.
  Use them only for local debugging. `.count()` is fine for
  monitoring or forcing materialization.
- `persist()` only when the DataFrame is reused. Pick
  `MEMORY_AND_DISK` by default; `MEMORY_ONLY` risks recomputation,
  `DISK_ONLY` is for very large frames.

## 8. Iceberg Write Patterns

- Use `.write.byName()` for schema evolution safety — Iceberg matches
  by name, not position.
- Read the latest partition via the `__partitions` metadata table.
  Don't full-scan to find the latest snapshot.
- Know your `write.distribution-mode`:
  - `"none"` — fastest, file sizes follow upstream partitioning
  - `"hash"` — re-shuffle by partition key, balanced file sizes
  - `"range"` — sort by key, expensive but ordered-scan friendly

Related

• Cursor Rules Guide | Python Claude Rules