{"filename":"pyspark-etl-best-practices-cursorrules-prompt-file.mdc","content":"---\ndescription: \"Cursor rules for PySpark ETL development with code style, joins, window functions, map operations, and Iceberg patterns.\"\nglobs: **/*\nalwaysApply: false\n---\nYou 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.\n\nFollow these rules when generating or reviewing PySpark code.\n\n# PySpark ETL Best Practices\n\n## 1. Project Structure\n\n### ETL class scaffold\n\nCreate 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.\n\n```python\nfrom abc import ABC, abstractmethod\nfrom pyspark.sql import SparkSession\n\nclass BaseETL(ABC):\n    def __init__(self, config, app_name=\"ETL Job\", spark_session=None):\n        self.spark = spark_session or SparkSession.builder.appName(app_name).getOrCreate()\n        self.config = config\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n    @abstractmethod\n    def run_job(self): ...\n\n    def stop(self):\n        self.spark.stop()\n```\n\n### Config — use a factory function\n\nKeep the dataclass as pure data and put CLI parsing in a standalone factory function. This makes configs easy to construct in tests without touching `sys.argv`.\n\n```python\n@dataclass\nclass MyConfig:\n    read_date: int = 20200101\n\ndef create_config() -> MyConfig:\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--read_date\", type=int, default=20200101)\n    args = parser.parse_args()\n    return MyConfig(read_date=args.read_date)\n```\n\n### Pipeline composition with `.transform()`\n\nKeep `run_job` as orchestration. Each step is a named method.\n\n```python\nevents = self.read_source().transform(self.enrich).transform(self.merge_with_existing)\n```\n\n### Use a shared reader for partition-aware reads\n\nBuild 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.\n\n```python\nclass PartitionedReader:\n    @staticmethod\n    def read_latest(spark, table_name, partition_col):\n        row = spark.read.table(table_name).agg(F.max(partition_col)).first()\n        if row is None or row[0] is None:\n            return spark.createDataFrame([], spark.read.table(table_name).schema)\n        return spark.read.table(table_name).filter(F.col(partition_col) == row[0])\n\n    @staticmethod\n    def read_by_date(spark, table_name, partition_col, date_value):\n        return spark.read.table(table_name).filter(F.col(partition_col) == date_value)\n\n# Reader handles partitioning\nevents = PartitionedReader.read_by_date(spark, \"catalog.my_table\", \"event_date\", 20260319)\n\n# Business filters stay in the ETL\nevents = events.filter(F.col(\"event_type\").isin(\"login\", \"purchase\"))\n```\n\n### Shared merge utilities\n\nFor 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).\n\n## 2. Code Style\n\n### Use `F.col()` — always use the `F.` prefix\n\nImport functions as `import pyspark.sql.functions as F` and use `F.col()`, `F.when()`, `F.lit()`, etc. throughout. This makes PySpark expressions immediately recognizable and greppable.\n\nAvoid `df.colA` attribute access — it binds the column to a specific DataFrame variable, which breaks after joins or when the variable is reassigned. Use `F.col()` with `.alias()` on the DataFrame if disambiguation is needed.\n\n```python\n# BAD — binds column to a specific DataFrame variable, breaks after joins\ndf.select(F.lower(df1.colA), F.upper(df2.colB))\n\n# GOOD\ndf.select(F.lower(F.col('colA')), F.upper(F.col('colB')))\n```\n\n### Extract complex conditions into named variables\n\nLimit logic inside `.filter()` or `F.when()` to 3 expressions. Extract the rest.\n\n```python\n# BAD — redundant logic hidden in nested parentheses\nF.when((F.col('status') == 'Delivered') | (((F.datediff('date_a', 'date_b') < 0) & ...)), 'Active')\n\n# GOOD\nis_delivered = (F.col('status') == 'Delivered')\ndate_passed = (F.datediff(F.col('date_a'), F.col('date_b')) < 0)\nhas_registration = (F.col('registration').rlike('.+'))\nF.when(is_delivered | (date_passed & has_registration), 'Active')\n```\n\n### Prefer `select` over `withColumn` chains\n\n`select` specifies the output schema in one pass. `withColumn` chains create intermediate DataFrames and can degrade performance — each call triggers a new projection in the query plan.\n\n```python\n# BAD — 3 intermediate DataFrames\ndf = df.withColumn(\"a\", F.col(\"a\").cast(\"double\"))\ndf = df.withColumn(\"b\", F.upper(F.col(\"b\")))\ndf = df.withColumn(\"c\", F.lit(1))\n\n# GOOD — 1 DataFrame, explicit schema contract\ndf = df.select(\n    F.col(\"a\").cast(\"double\"),\n    F.upper(F.col(\"b\")).alias(\"b\"),\n    F.lit(1).alias(\"c\"),\n)\n```\n\n### Use `alias` over `withColumnRenamed`\n\n```python\n# BAD\ndf.select('key', 'comments').withColumnRenamed('comments', 'num_comments')\n\n# GOOD\ndf.select('key', F.col('comments').alias('num_comments'))\n```\n\n### Chaining limits\n\nMax 5 statements per chain. Separate by operation type (select/filter vs withColumn vs join).\n\n```python\n# BAD — mixed concerns in one chain\ndf = (df.select('a', 'b', 'key')\n    .filter(F.col('a') == 'x')\n    .withColumn('ratio', F.col('a') / F.col('b'))\n    .join(df2, 'key', how='inner')\n    .drop('b'))\n\n# GOOD — separated by concern\ndf = df.select('a', 'b', 'key').filter(F.col('a') == 'x')\ndf = df.withColumn('ratio', F.col('a') / F.col('b'))\ndf = df.join(df2, 'key', how='inner').drop('b')\n```\n\n## 3. Joins\n\n### Always specify `how=` explicitly\n\n```python\n# BAD\ndf.join(other, 'key')\n\n# GOOD\ndf.join(other, 'key', how='inner')\n```\n\n### Prefer left joins over right joins\n\nFlip the DataFrame order and use `left` instead of `right` for readability — the primary DataFrame stays on the left.\n\n```python\nflights = flights.join(aircraft, 'aircraft_id', how='left')\n```\n\n### Use `.alias()` for disambiguation after joins\n\n```python\n# BAD — renaming every column before join\nfor c in columns:\n    flights = flights.withColumnRenamed(c, 'flights_' + c)\n\n# GOOD — alias the whole DataFrame\nflights = flights.alias('f')\nparking = parking.alias('p')\nresult = flights.join(parking, 'code', how='left').select(\n    F.col('f.start_time').alias('flight_start'),\n    F.col('p.total_time').alias('parking_total'),\n)\n```\n\n### Broadcast small dimension tables\n\nWhen joining a large fact DataFrame with a small lookup/dimension table, wrap the small side in `F.broadcast()` to skip the shuffle on the small side.\n\nUse broadcast for tables that are small enough to fit in executor memory — typically dimension/lookup tables (category lookups, country codes, config mappings). Spark auto-broadcasts tables under 10MB by default (`spark.sql.autoBroadcastJoinThreshold`), but an explicit hint is useful when Spark can't infer the size (e.g., after filters or transformations).\n\nTo check if a table is broadcast-worthy during development:\n- **Spark UI**: after a run, check the SQL tab — scan sizes are shown per table\n- **Quick row count**: `spark.read.table(\"catalog.my_dim\").count()` (dev only, not in production code)\n- **Query plan**: `df.explain()` — Spark shows `BroadcastHashJoin` if it auto-broadcasts, `SortMergeJoin` if it doesn't\n\n```python\ndf.join(F.broadcast(category_dim), 'category_id', how='left')\n```\n\n### Never use `.dropDuplicates()` as a crutch\n\nIf duplicate rows appear, find the root cause. `.dropDuplicates()` masks the problem and adds shuffle overhead.\n\n## 4. Window Functions\n\nUse `from pyspark.sql import Window as W` alongside `import pyspark.sql.functions as F`.\n\n### Always specify an explicit frame\n\nWithout a frame, Spark picks one that changes depending on whether `orderBy` is present.\n\n```python\n# BAD — F.sum gives running sum with orderBy, total without. Surprising.\nw = W.partitionBy('key').orderBy('num')\n\n# GOOD — explicit about what you want\nw = W.partitionBy('key').orderBy('num').rowsBetween(W.unboundedPreceding, W.unboundedFollowing)\n```\n\n### `row_number` + filter vs `first` — know the difference\n\n- `row_number` + filter = **drop rows**, keep the best one\n- `first` over window = **overwrite a column value**, keep all rows\n\n### Use `ignorenulls=True` with `first` and `last`\n\nWithout it, a null in the first row gives null for the entire partition.\n\n```python\n# BAD — returns None if first row is null\nF.first('version').over(window)\n\n# GOOD\nF.first('version', ignorenulls=True).over(window)\n```\n\n### Avoid empty `partitionBy()`\n\nIt forces all data into one partition. Use `.agg()` instead for global aggregations.\n\n```python\n# BAD — single partition, kills performance\nw = W.partitionBy()\ndf.select(F.sum('num').over(w))\n\n# GOOD\ndf.agg(F.sum('num').alias('total'))\n```\n\n## 5. Map & Array Higher-Order Functions\n\n### Use `map_zip_with` when merging maps with complex logic\n\n`map_concat` is fine for simple merges with no key overlap. When you need custom logic per key (e.g., keep the newer timestamp, pick the higher value), use `map_zip_with` — it gives you a per-key merge function instead of blindly letting one side win.\n\n```python\n# BAD — no control over conflict resolution\nmap_concat(existing_map, new_map)\n\n# GOOD — keep the entry with the later timestamp\nmap_zip_with(new_map, existing_map,\n    lambda key, v1, v2: (\n        F.when(v1.isNull(), v2)\n        .when(v2.isNull(), v1)\n        .otherwise(F.when(v1.event_ts >= v2.event_ts, v1).otherwise(v2))\n    )\n)\n```\n\n### Use `transform` + `array_max` to extract from nested structs\n\n```python\n# Extract max event_ts from a map of structs\narray_max(transform(map_values(my_map), lambda x: x.event_ts))\n```\n\n### Avoid UDFs — use native Spark functions first\n\nUDFs break Catalyst optimization and add serialization overhead. Before writing one, check if a built-in Spark function or higher-order function can do the job.\n\n## 6. Cumulative / Snapshot Table Patterns\n\n### Merges must be idempotent\n\nRe-running with the same data should produce the same result, not create duplicates.\n\n### Merges must be order-independent\n\nBackfilling old data should not overwrite newer data. Use an explicit ordering criterion (e.g., event timestamp, version number, partition date) to resolve conflicts — don't rely on positional precedence like `coalesce` argument order.\n\n### Validate primary key uniqueness after writes\n\nAdd audit steps that validate primary key uniqueness and check for nulls in key columns.\n\n## 7. Data Quality & Performance\n\n### Use `F.lit(None)` for empty columns, never empty strings\n\n```python\n# BAD\ndf = df.withColumn('foo', F.lit(''))\ndf = df.withColumn('foo', F.lit('NA'))\n\n# GOOD\ndf = df.withColumn('foo', F.lit(None))\n```\n\n### Avoid `.otherwise()` as a general catch-all\n\nUnknown values silently collapse into the otherwise bucket, hiding data quality issues.\n\n```python\n# BAD — a new platform_type you didn't anticipate becomes \"Other\" silently\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n .otherwise('Other')\n\n# GOOD — unmapped values stay null, surfacing gaps in your logic\nF.when(F.col('platform_type') == 'android', 'Mobile')\n .when(F.col('platform_type') == 'ios', 'Mobile')\n```\n\n### No `.show()`, `.collect()`, `.printSchema()` in production\n\nThese trigger full materialization or add unnecessary driver overhead. Use them only for local debugging, never in deployed ETL code. `.count()` is acceptable when used intentionally (e.g., logging row counts for monitoring, forcing materialization before a DAG fork).\n\n### Use `persist()` intentionally\n\nOnly persist a DataFrame when it's referenced in multiple subsequent actions — otherwise the write action will materialize it for you. `.persist()` + `.count()` is a common pattern to force materialization and log row counts for debugging; use it when needed but be aware it adds a full scan.\n\nChoose the storage level based on your use case:\n- `MEMORY_AND_DISK` — safe default, spills to disk if memory is tight\n- `MEMORY_ONLY` — faster but risks recomputation if evicted\n- `DISK_ONLY` — for very large DataFrames that don't fit in memory\n\n## 8. Iceberg Write Patterns\n\n### Use `.byName()` for schema evolution safety\n\nColumn ordering doesn't matter — Spark matches by name, not position.\n\n```python\ndf.write.byName().mode(\"overwrite\").insertInto(\"catalog.my_table\")\n```\n\n### Use `__partitions` metadata table for latest partition reads\n\nIceberg exposes a `__partitions` metadata table. Use it to find the latest snapshot instead of scanning the full table.\n\n```python\npartition_df = spark.read.table(\"catalog.my_table__partitions\").select(\n    \"partition.partition_date\", \"partition.partition_hour\"\n)\nmax_partition = partition_df.orderBy(\n    F.col(\"partition_date\").desc(), F.col(\"partition_hour\").desc()\n).first()\nif max_partition is None:\n    raise ValueError(\"No partitions found in catalog.my_table\")\nlatest_date = max_partition[\"partition_date\"]\n```\n\n### Understand `write.distribution-mode`\n\n- `\"none\"` — no re-shuffle before writing. Fastest, but output file sizes depend on upstream partitioning.\n- `\"hash\"` — redistributes data by partition key. Produces evenly sized files but adds a shuffle.\n- `\"range\"` — sorts data by partition key before writing. Good for ordered scan performance but most expensive.\n"}