pyspark-expert

name	pyspark-expert
description	Produces production-quality PySpark code, unit tests, and project layouts for Spark 3.5+ with Delta Lake. Use when the user asks to write, review, refactor, or debug PySpark code, define schemas, design ETL modules, or discuss DataFrame API best practices. Also trigger for: PySpark, Spark DataFrame, SparkSession, pyspark.sql.functions, Delta Lake, MERGE INTO, time travel, OPTIMIZE, ZORDER, VACUUM, Change Data Feed, readStream, writeStream, Structured Streaming, foreachBatch, or assertDataFrameEqual. Do NOT use for generic Python questions, non-Spark data tools (pandas, Polars, DuckDB), or writing pytest tests (use pytest-data-engineering for that).
allowed-tools	Read Write Edit Bash Grep
license	GPL-3.0-only
status	unstable
metadata	{"author":"Leandro Kellermann de Oliveira <[e-mail]>","version":"0.1.0","model":"sonnet"}

PySpark Expert

This skill produces production-quality PySpark code targeting Spark 3.5+ with Delta Lake. It enforces a consistent project layout, idiomatic DataFrame API usage, DDL-based schema definitions, and pytest patterns with separated test data files.

Consult the routing table at the end of this file to choose which reference(s) to read before writing or reviewing code.

---

Core Principles

1. Functional composition over mutation chains — prefer df.select(...) and df.transform(fn) over long .withColumn() chains.
2. Schemas are contracts — define every schema as a DDL string constant, never inline StructType builders. The same schema constants are used in production code and in test data fixtures.
3. Built-ins over UDFs — exhaust pyspark.sql.functions and pyspark.sql.Column methods before reaching for a UDF. When a UDF is genuinely needed, explain why no built-in alternative exists.
4. Explicit over implicit — always alias derived columns, qualify ambiguous column references in joins, and specify column lists in select() rather than select("*").
5. Test with real Spark — no mocking of DataFrame internals. Use a shared SparkSession pytest fixture and assertDataFrameEqual which validates both schema and data in one assertion.

---

Project Layout (enforced)

When generating a PySpark project or module, always use this structure:

project_name/
├── src/
│   ├── __init__.py
│   ├── schemas.py             # DDL schema constants (single source of truth)
│   ├── transformations.py     # Pure DataFrame → DataFrame functions
│   ├── readers.py             # I/O: read from sources
│   ├── writers.py             # I/O: write to sinks
│   └── tasks/                 # Entry points that compose read → transform → write
│       ├── __init__.py
│       └── process_orders.py
├── tests/
│   └── unit_tests/
│       ├── conftest.py                # Shared SparkSession fixture
│       ├── transformations_data.py    # Mock inputs & expected outputs as fixtures
│       └── transformations_test.py    # Flat pytest functions
├── pyproject.toml
└── README.md

The separation matters because:

• schemas.py is the single source of truth — imported by both production code and test data fixtures. Never redefine schemas in tests.
• transformations.py contains pure functions (DataFrame) → DataFrame with no I/O, which makes them trivially testable.
• readers.py / writers.py isolate I/O so tests never touch real storage.
• tasks/ are thin orchestrators; they hold no business logic.
• _data.py / _test.py pairs keep mock data separate from test logic.

For larger projects, schemas.py and transformations.py can become packages (schemas/, transformations/) with submodules. The test file naming convention still applies: one <module>_data.py and one <module>_test.py per source module.

---

Schema Definition Rules

Always define schemas as DDL string constants in src/schemas.py — never inline StructType builders. Use the same constants in production code, test fixtures, and CREATE TABLE statements. See references/transformations.md SP-01 for the canonical example and rationale.

---

Code Style Rules

See references/transformations.md and references/anti-patterns.md for detailed patterns and examples:

• Prefer select() + transform() over .withColumn() chains — one projection, not N sequential Project nodes in the logical plan (AP-01).
• Use DataFrame.transform() for pipeline composition — chain pure functions cleanly (TP-01, TP-02).
• Never use collect() unless you genuinely need all data on the driver (AP-02).
• Always qualify join columns to avoid ambiguity (AP-04, JP-01).
• Import style: import pyspark.sql.functions as F / import pyspark.sql.types as T. Never from pyspark.sql.functions import * (AP-08).

---

TDD workflow (enforced)

When developing a new transformation, use a Python virtual environment and always follow this order:

1. Define the mock input and expected output fixtures in _data.py.
2. Write the test function in _test.py — it will fail because the transformation doesn't exist yet.
3. Implement the transformation in src/transformations.py until the test passes.
4. Add edge cases (nulls, empty DataFrames, constraint violations) as new fixture + test pairs.

Use @pytest.mark.skip(reason="TBD") to stub tests you plan to implement later or if you failed to implement more than 3 times. In case of failure, write a TODO.md file in the project root folder with this structure:

# TODO

## Skipped Tests

| Test       | File                                       | Reason               | Attempted approaches |
| ---------- | ------------------------------------------ | -------------------- | -------------------- |
| `test_foo` | `tests/unit_tests/transformations_test.py` | <why it was skipped> | <what was tried>     |

## Known Blockers

- <dependency version conflict, missing built-in, etc.>

---

Spark 3.5+ Features to Prefer

When writing code for Spark 3.5+, use these modern APIs:

Instead of	Use (3.5+)
Manual try/except on casts	F.try_to_number(), F.try_cast()
.withColumn() chains	.select() + .transform()
collect_list without ordering	collect_list + .orderBy() in window
Multiple when/otherwise for ranges	F.greatest(), F.least() when applicable
Manual null-safe comparisons	Column.eqNullSafe() (<=>)
Row-level Python UDFs	pandas_udf for transforms, foreachPartition for side effects
Indiscriminate .cache()	Cache only DataFrames reused 2+ times; always unpersist() when done
Python print() in jobs	log4j via spark.sparkContext._jvm (see references/jobs-and-tuning.md LG-01)
trigger(once=True) in Streaming	trigger(availableNow=True) (parallel micro-batches, cleaner stop)

---

Minimal documentation

Always consider the following README.md content when working on a project:

• About: what is the project about.
• Getting Started: commands to install, configure, run unit tests with pytest, and how is expected to run.
• IMPROVEMENTS.md: file summarizing main improvements of the new refactored version in comparison with the original one.

---

Quick Reference: pytest patterns

For test file layout, fixture conventions, conftest, pytest.toml, .coveragerc, ruff.toml, and CI configuration, use the pytest-data-engineering skill — it is the canonical source for all pytest patterns in this repository.

---

Reference Routing Table

Read only the file(s) needed for the current task.

When the task is…	Read
Writing or refactoring a transformation, join, aggregation, schema, or UDF decision	references/transformations.md + references/anti-patterns.md
Reviewing existing PySpark code for problems	references/anti-patterns.md
Caching, partitioning, file sizing, SparkSession config, or logging	references/jobs-and-tuning.md
Structured Streaming pipelines	references/streaming.md
Delta tables, MERGE, time travel, OPTIMIZE, VACUUM, or Change Data Feed	references/DELTA_PATTERNS.md