How do I write a Spark DataFrame partitioned by a column?

Summary

• Call partitionBy() on the DataFrame writer, for example df.write.partitionBy("sale_date").parquet(path). Spark lays the output out Hive-style, with one sale_date=<value> directory per distinct value.
• Partition values live in the directory path. When you read the table root, Spark's partition discovery turns them back into columns and infers numeric, date, timestamp and string types by default.
• To rewrite only the partitions present in new data, set the writer option partitionOverwriteMode to dynamic. The default mode, set by spark.sql.sources.partitionOverwriteMode, is STATIC.
• Repartition on the partition column before writing so each value's rows are written together, and cap rows per file with spark.sql.files.maxRecordsPerFile (default 0, no limit).
• Partition only on low-cardinality columns that queries filter on. Spark's docs say partitionBy has limited applicability to high-cardinality columns and point to bucketBy for unbounded values.

How do I write a Spark DataFrame partitioned by a column?

Use DataFrameWriter.partitionBy(). It has been in PySpark since version 1.4 and works with save(), saveAsTable() and format shortcuts such as .parquet(). Spark writes one subdirectory per distinct value, named column=value, and every built-in file source (Parquet, ORC, JSON, CSV and text) can discover that layout when it reads the root path. The call itself is one line. Picking the column is the real decision: a good one lets queries that filter on it read only matching partitions, and a bad one leaves the table full of tiny files. Choose a low-cardinality column that queries filter on, such as a date.

Write the table

sales = spark.read.parquet("/raw/sales")

(sales.repartition("sale_date")
      .write
      .partitionBy("sale_date")
      .mode("overwrite")
      .parquet("/lake/sales"))

june_first = spark.read.parquet("/lake/sales").where("sale_date = '2026-06-01'")

The output has one directory per date, such as /lake/sales/sale_date=2026-06-01/. The repartition("sale_date") call hash-partitions rows by date before the write, so each date's rows sit in one task and are written together instead of being scattered across every task's output.
To create a persistent, partitioned table in SQL instead:

CREATE TABLE sales_by_date
USING parquet
PARTITIONED BY (sale_date)
AS SELECT * FROM parquet.`/raw/sales`;

Overwrite only the partitions you changed

In the default static mode, mode("overwrite") replaces the existing data with the DataFrame's contents, so a daily job that writes one date would remove all the others. Dynamic mode deletes nothing ahead of time and overwrites only the partitions that receive rows:

(daily_sales.write
    .mode("overwrite")
    .option("partitionOverwriteMode", "dynamic")
    .partitionBy("sale_date")
    .parquet("/lake/sales"))

The writer option takes precedence over the session setting spark.sql.sources.partitionOverwriteMode. In SQL, INSERT OVERWRITE with a PARTITION (col = value) clause targets one named partition.

Options that shape the output

Option	When to use it	Trade-off
partitionBy("col")	Queries filter on a low-cardinality column such as a date	Every distinct value becomes a directory, so high cardinality means many small files
repartition("col") before the write	You want fewer, larger files in each directory	Adds a shuffle, and one very large value is written by a single task
spark.sql.files.maxRecordsPerFile	A partition is big enough to need several files	Default 0 means no limit
REBALANCE hint (SQL)	SQL writes that should avoid too-small and too-big files	Best effort, and ignored when adaptive query execution is off
partitionOverwriteMode set to dynamic	Reprocessing some partitions without touching the rest	Hive SerDe tables ignore the setting and always overwrite dynamically
bucketBy(n, "col") with saveAsTable()	High-cardinality keys you join or filter on	Persistent tables only, and not compatible with Hive's bucketing

Read the data back

Point the reader at the table root, not at a single directory. Spark extracts sale_date from the paths and adds it to the schema. If you read one partition directory on its own, sale_date isn't treated as a column unless you set the basePath option to the table root. Set spark.sql.sources.partitionColumnTypeInference.enabled to false if you want partition values as strings. For persistent tables, Spark keeps per-partition metadata in the metastore, so a query only asks for the partitions it needs. External tables created with a path option don't gather that information by default; run MSCK REPAIR TABLE to sync it.

Delta Lake and Apache Iceberg tables

Delta Lake uses the same writer call: df.write.format("delta").partitionBy("sale_date").saveAsTable("sales"). For selective rewrites, the Delta docs recommend replaceWhere or, in Delta Lake 4.3.0 and above, replaceUsing, and they list partitionOverwriteMode as legacy for Delta tables since 4.3.0. Don't read a Delta table's partition directories directly; Hive-style partitioning isn't part of the Delta Lake protocol. Apache Iceberg tables take partition transforms such as day(ts) or bucket(N, col) in PARTITIONED BY, so readers don't have to filter on a derived column.

How Databricks approaches partitioned writes

Databricks recommends liquid clustering for all managed tables: declare CLUSTER BY (sale_date) instead of PARTITIONED BY, or CLUSTER BY AUTO on Unity Catalog managed tables to let predictive optimization pick and adjust the keys. The Databricks partitioning guide advises against partitioning tables with less than 1 TB of data and says each partition should hold at least 1 GB. Unpartitioned Delta tables are clustered by ingestion time automatically. To move an existing partitioned Delta table, Databricks Runtime 18.1 and above supports ALTER TABLE ... REPLACE PARTITIONED BY WITH CLUSTER BY.

FAQs

Why is the partition column missing when I read one folder?

The value is encoded in the directory path, and Spark only adds it as a column when it discovers partitions from the table root. Read the root, or set basePath to the root when you read a subdirectory.

Can I partition by more than one column?

Yes. partitionBy("country", "sale_date") nests the directories, for example country=FR/sale_date=2026-06-01/. Each extra column multiplies the number of directories, so keep the combination coarse.

Is partitionBy the same as repartition?

No. repartition() controls how rows are split across tasks inside the job, while partitionBy() controls the directory layout in storage. They're often used together, as in the first example.