Pyspark Etl Best Practices (.cursorrules)

{
    "prompt": "You are an expert in PySpark, Spark SQL, Apache Iceberg, and production data engineering. You write performant, idiomatic ETL code that is testable, readable, and safe for cumulative/snapshot tables. # PySpark ETL Best Practices ## 1. Project Structure ### ETL class scaffold Create a base class that manages the SparkSession lifecycle. Accept an optional `spark_session` parameter so tests can inject a local session. Use an abstract method for the job logic. ```python from abc import ABC, abstractmethod from import SparkSession class BaseETL(ABC): def __init__(self, config, app_name=\"ETL Job\", spark_session=None): = spark_session or e(app_name).getOrCreate() = config = ger(self.__class__.__name__) def run_job(self):. def stop(self): () ``` ### Config — use a factory function Keep the dataclass as pure data and put CLI parsing in a standalone factory function. This makes configs easy to construct in tests without touching ``. ```python class MyConfig: read_date: int = 20200101 def create_config() -> MyConfig: parser = ntParser() _argument(\"--read_date\", type=int, default=20200101) args = _args() return MyConfig(read_date=_date) ``` ### Pipeline composition with `.transform()` Keep `run_job` as orchestration. Each step is a named method. ```python events = _source().transform().transform(_with_existing) ``` ### Use a shared reader for partition-aware reads Build a generic reader utility that handles partition mechanics (date filters, hour ranges, latest-partition lookups). Don't create one-off reader classes per table — keep domain-specific filters in the ETL where they're visible. ```python class PartitionedReader: def read_latest(spark, table_name, partition_col): row = (table_name).agg((partition_col)).first() if row is None or row[0] is None: return DataFrame([], (table_name).schema) return (table_name).filter((partition_col) == row[0]) def read_by_date(spark, table_name, partition_col, date_value): return (table_name).filter((partition_col) == date_value) # Reader handles partitioning events = _by_date(spark, \"_table\", \"event_date\", 20260319) # Business filters stay in the ETL events = ((\"event_type\").isin(\"login\", \"purchase\")) ``` ### Shared merge utilities For simple outer-join-with-coalesce merges, build a reusable merge function that handles aliasing, join key coalescing, and per-column defaults. Use `map_zip_with` when you need per-key conflict resolution (timestamp-aware merges). ## 2. Code Style ### Use `()` — always use the `F.` prefix Import functions as `import ons as F` and use `()`, `()`, `()`, etc. throughout. This makes PySpark expressions immediately recognizable and greppable. Avoid `` attribute access — it binds the column to a specific DataFrame variable, which breaks after joins or when the variable is reassigned. Use `()` with `.alias()` on the DataFrame if disambiguation i",
    "model": "Claude 3.7 Sonnet / Cursor",
    "aspect_ratio": "1:1",
    "category": "Coding Agents",
    "content_type": "coding_prompt",
    "surface": "agents_hub",
    "quality_score": 81,
    "provenance": "SYSTEM_RULE"
}