
You've built a pipeline that lands 50 million rows of order data into S3 every night. Analysts love the data. But every morning, the first query of the day takes four minutes to run — scanning every single file, cold, across the entire dataset. Your data warehouse is billing you by the byte scanned. Your BI tool times out. Someone opens a Slack thread that starts with "the data pipeline is broken" when really, the data is fine — it's just organized poorly.
Partitioning is the structural decision that separates a dataset that works from a dataset that performs. When you write data out of a pipeline, you're not just saving records — you're making a bet about how that data will be read. A well-partitioned output means downstream queries can skip irrelevant data entirely, sometimes reducing scan volume by 99%. A poorly partitioned output means every query does a full scan regardless of how targeted it is. The write side of your pipeline shapes the read performance for everyone downstream, often for months or years.
By the end of this lesson, you'll understand the mechanical reality of how partitioning affects query planning, and you'll be able to make deliberate choices between date, hash, and range partitioning strategies — and combinations of them — based on actual query patterns.
What you'll learn:
You should be comfortable with:
WHERE clauses and GROUP BYBefore choosing a strategy, you need to have the right mental model. Partitioning in pipeline output is not a database index. It's a naming convention for the directories where files land.
When you write a partitioned Parquet dataset to S3, the output looks like this:
s3://data-lake/orders/
year=2024/month=01/day=15/part-00000.parquet
year=2024/month=01/day=16/part-00000.parquet
year=2024/month=01/day=17/part-00000.parquet
year=2024/month=02/day=01/part-00000.parquet
When a query engine like Athena, Spark SQL, Presto, or BigQuery's external tables reads this dataset, it first lists the directory structure. If your query contains WHERE event_date = '2024-01-15', the engine maps that filter to the directory path year=2024/month=01/day=15/ and never opens the files in any other directory. This is called partition pruning, and it happens before any data is read.
The key insight: partition pruning eliminates file I/O entirely. It's not that the engine reads the files and discards non-matching rows — it skips the files altogether. A query that filters on a well-partitioned column doesn't scan 50 million rows; it might scan 150,000.
Here's a minimal PySpark example to build intuition before we go deeper:
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date
spark = SparkSession.builder.appName("partition-demo").getOrCreate()
# Load raw orders from your pipeline's staging zone
orders_df = spark.read.parquet("s3://data-lake/raw/orders/")
# Normalize your partition column before writing
orders_df = orders_df.withColumn(
"order_date", to_date(col("created_at"))
)
# Write partitioned output — this creates the directory hierarchy
orders_df.write \
.mode("overwrite") \
.partitionBy("order_date") \
.parquet("s3://data-lake/processed/orders/")
After this write, a query like SELECT * FROM orders WHERE order_date = '2024-03-22' will only read the files under order_date=2024-03-22/. The engine doesn't touch anything else.
Warning: The query engine must be made aware of the partition structure. In Hive-metastore-backed systems like Glue, you need to run
MSCK REPAIR TABLEor useALTER TABLE ADD PARTITIONafter landing new partitions. Athena and Spark SQL can also useREFRESH TABLE. If you skip this step, queries will ignore new partitions entirely — and your pipeline will appear to deliver no data.
Date partitioning is the most common strategy because most analytical workloads are time-bounded. Reports are almost always scoped to a date range: "last 30 days," "Q3," "yesterday's orders." This makes time a natural partitioning key.
The decision isn't just "partition by date" — it's "partition by what level of date granularity." Your options are year, month, day, and hour, and each has a different trade-off between partition count and file size.
Consider an e-commerce platform processing 2 million events per day. If you partition at the day level, each partition holds 2 million rows, which compresses to roughly 200MB of Parquet — a good size for most engines. If you partition at the hour level, you get 24 partitions per day, each with ~83,000 rows and ~8MB. That sounds fine until you realize 90 days of hourly data is 2,160 directories and potentially 2,160+ files. Listing that directory structure has real overhead, and many query engines perform poorly with thousands of tiny files.
The small-file problem is the most common mistake with time partitioning. Here's the pattern that causes it:
# DON'T DO THIS on a modest event stream
events_df.write \
.mode("append") \
.partitionBy("event_year", "event_month", "event_day", "event_hour") \
.parquet("s3://data-lake/events/")
If your pipeline runs every 15 minutes and appends to hourly partitions, after a week you have hundreds of tiny files scattered across hourly directories. Each Spark task that reads a partition has to open, deserialize, and close dozens of files. The fix is to either coalesce before writing or compact files periodically:
from pyspark.sql.functions import year, month, dayofmonth, hour
events_df = events_df \
.withColumn("event_year", year(col("event_ts"))) \
.withColumn("event_month", month(col("event_ts"))) \
.withColumn("event_day", dayofmonth(col("event_ts"))) \
.withColumn("event_hour", hour(col("event_ts")))
# Repartition to control the number of output files per partition
# Each output partition here will produce exactly 1 file
events_df \
.repartition(1, "event_year", "event_month", "event_day", "event_hour") \
.write \
.mode("overwrite") \
.partitionBy("event_year", "event_month", "event_day", "event_hour") \
.parquet("s3://data-lake/events/")
Tip: The
repartition(1, ...)call groups all rows with the same partition key values into a single Spark partition, which writes a single file. Use this when you want exactly one file per output directory. For very large partitions, use a larger number (e.g.,repartition(4, "event_year", "event_month", "event_day")) to keep individual files under 256MB.
One subtle issue: when you use partitionBy, Spark removes those columns from the Parquet file contents. The column values are encoded in the directory path only. Most query engines reconstruct them automatically when you read back. But if you need the column to be both a partition key and present in the file body — for compatibility with tools that read files directly without a metastore — write the column twice with different names, or verify your reader reconstructs it correctly.
Real pipelines deal with late data. An event timestamped yesterday may arrive in your pipeline tomorrow. If you always partition by processing date, late events land in today's partition with yesterday's timestamps. If you partition by event date, you need to overwrite or append to old partitions.
A practical pattern for late data with Spark:
# Use "event_date" derived from the event timestamp, not from processing time
orders_enriched = orders_enriched.withColumn(
"event_date", to_date(col("order_placed_at"))
)
# Write with dynamic partition overwrite — only overwrites partitions
# that appear in this batch, leaving other partitions untouched
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
orders_enriched.write \
.mode("overwrite") \
.partitionBy("event_date") \
.parquet("s3://data-lake/processed/orders/")
With partitionOverwriteMode set to dynamic, Spark only overwrites the specific date partitions present in the current DataFrame. A batch containing event_date = 2024-03-20 and event_date = 2024-03-22 will overwrite only those two partitions, leaving event_date = 2024-03-21 intact. This is the correct behavior for reprocessing pipelines.
Date partitioning is intuitive, but it has a structural weakness: it doesn't distribute data evenly across partitions when you're not filtering by time. If you're running a pipeline that feeds a system where queries join large tables on customer_id, a date-partitioned layout forces those joins to shuffle data across every partition. Hash partitioning solves a different problem than date partitioning — it's about co-location and uniformity, not temporal filtering.
Hash partitioning applies a hash function to a column value and uses the result modulo N (the number of buckets) to assign rows to output files. The key property is that the same value always maps to the same bucket. Two tables hash-partitioned on customer_id with the same number of buckets will place the same customer's rows in corresponding files in both tables.
# Hash-partition orders and customer_events on customer_id
# Using the same number of partitions is critical for co-location
NUM_BUCKETS = 64 # Choose based on data volume
orders_df \
.repartition(NUM_BUCKETS, col("customer_id")) \
.write \
.mode("overwrite") \
.parquet("s3://data-lake/processed/orders_hashed/")
customer_events_df \
.repartition(NUM_BUCKETS, col("customer_id")) \
.write \
.mode("overwrite") \
.parquet("s3://data-lake/processed/events_hashed/")
When a Spark job reads these two datasets and joins them on customer_id, it can perform a sort-merge join without a full shuffle — because corresponding buckets already contain matching keys. This eliminates what's often the most expensive operation in large-scale joins.
Warning: In Spark's standard
repartition(), the output directories don't carry partition column names in their paths (unlikepartitionBy()). You get files namedpart-00000.parquetthroughpart-00063.parquetin a flat directory. Partition pruning based on column values doesn't apply here — the benefit is in-memory shuffle reduction during joins, not scan reduction during filtering. If you need both, combine hash with a higher-level date partition (covered later in Composite Partitioning).
Getting the bucket count wrong is the most consequential decision in hash partitioning. Too few buckets and each file becomes enormous; too many and you're back to the small-file problem.
A reasonable rule of thumb: target files of 128MB–512MB when compressed. Work backward from your data volume:
Total uncompressed data size: 500 GB
Parquet compression ratio: ~5x
Compressed size: ~100 GB
Target file size: 256 MB
Number of buckets: 100 GB / 256 MB ≈ 400 buckets
But there's another constraint: bucket counts that are powers of 2 (64, 128, 256) make future bucket coalescing easier. If your dataset grows and you later want to merge 256 buckets into 128, buckets that are powers of 2 divide cleanly.
Tip: For Delta Lake users, Hive-style bucketing via
bucketBy()encodes bucket information in the table metadata, enabling automatic bucket join optimizations without manual co-location management. This is preferable to manualrepartition()when your lakehouse format supports it.
Use hash partitioning when:
It's the wrong choice when:
WHERE date BETWEEN x AND yRange partitioning divides data into contiguous value ranges rather than hashing or temporal slices. Think of it as a way to encode sorted order into your directory structure. It's the right strategy when your query predicates involve ordered comparisons: WHERE revenue BETWEEN 1000 AND 5000, WHERE customer_id BETWEEN 100000 AND 200000, or tiered access patterns like pricing bands.
Unlike date partitioning (which uses a natural column) and hash partitioning (which is computed mathematically), range partitioning requires you to define the boundaries explicitly. You're essentially building a bucketing scheme based on domain knowledge.
from pyspark.sql.functions import when, col
# Partition a product catalog by price range
# Boundaries defined based on business knowledge of price distribution
products_df = products_df.withColumn(
"price_tier",
when(col("price") < 10, "tier_budget")
.when(col("price") < 50, "tier_economy")
.when(col("price") < 200, "tier_standard")
.when(col("price") < 1000, "tier_premium")
.otherwise("tier_luxury")
)
products_df.write \
.mode("overwrite") \
.partitionBy("price_tier") \
.parquet("s3://data-lake/processed/products/")
A query like SELECT * FROM products WHERE price_tier = 'tier_standard' now scans only the tier_standard/ directory. But more importantly, a query like SELECT * FROM products WHERE price BETWEEN 50 AND 200 can be pushed down to the same partition with minimal scanning — if your query engine supports partition predicate pushdown for derived columns.
Range partitioning is particularly valuable when you have a large table accessed by ID ranges — customer IDs, transaction IDs, product SKUs. The classic use case is when customer service tooling needs to look up records for a segment of customers, or when you're running a pipeline that processes customers in ID batches.
# Determine approximate quantile boundaries for customer_id
# This is important: use actual data distribution, not guesswork
quantiles = customers_df.approxQuantile("customer_id", [0.25, 0.5, 0.75], 0.01)
q1, q2, q3 = quantiles
customers_df = customers_df.withColumn(
"customer_shard",
when(col("customer_id") < q1, "shard_a")
.when(col("customer_id") < q2, "shard_b")
.when(col("customer_id") < q3, "shard_c")
.otherwise("shard_d")
)
customers_df.write \
.mode("overwrite") \
.partitionBy("customer_shard") \
.parquet("s3://data-lake/processed/customers/")
Using approxQuantile ensures your partitions are balanced by row count, not just by value range. The difference matters: if 60% of your customers have IDs above 500,000, an evenly spaced split at 250K/500K/750K would create a massively skewed partition.
Range partitioning's greatest failure mode is skew — boundaries that look reasonable but produce wildly uneven partitions. Before finalizing your boundaries, inspect the distribution:
from pyspark.sql.functions import count, min as spark_min, max as spark_max
# After writing, check partition statistics to verify balance
partitioned_customers = spark.read.parquet("s3://data-lake/processed/customers/")
partitioned_customers.groupBy("customer_shard") \
.agg(
count("*").alias("row_count"),
spark_min("customer_id").alias("id_min"),
spark_max("customer_id").alias("id_max")
) \
.orderBy("customer_shard") \
.show()
If one shard has 10x the rows of another, your boundaries need to be adjusted. Rerun with updated quantile thresholds and rewrite the dataset.
Production data pipelines rarely serve a single query pattern. An orders table might be queried by date range 70% of the time, but 30% of queries filter on region and then join on order_id. A single partitioning strategy optimizes for one pattern and ignores the others. Composite partitioning layers multiple strategies to cover multiple access patterns.
The most common composite strategy is a temporal outer partition with hash inner bucketing. The outer partition provides pruning for time-range queries; the inner hash distribution optimizes joins within a time window.
from pyspark.sql.functions import to_date, col
NUM_BUCKETS_PER_DAY = 16 # Sized for daily partition volume
orders_enriched = orders_enriched \
.withColumn("order_date", to_date(col("created_at")))
# Step 1: Partition by date at the directory level
# Step 2: Within each date partition, hash by customer_id for join efficiency
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
orders_enriched \
.repartition(NUM_BUCKETS_PER_DAY, col("order_date"), col("customer_id")) \
.write \
.mode("overwrite") \
.partitionBy("order_date") \
.parquet("s3://data-lake/processed/orders_composite/")
Here's what this achieves:
order_date reads only one date's directoryNUM_BUCKETS_PER_DAY files exist, hash-distributed on customer_idcustomer_id-repartitioned dataset within the same date window can skip inter-partition shufflesWhen you serve multiple teams with different SLAs, you might want to combine date and range to create a tiered layout:
# Combine temporal partitioning with revenue range partitioning
# Useful when analysts frequently run queries like:
# "Show me high-value orders from the past 30 days"
transactions_df = transactions_df \
.withColumn("tx_date", to_date(col("transaction_time"))) \
.withColumn(
"value_tier",
when(col("transaction_amount") < 100, "low")
.when(col("transaction_amount") < 1000, "medium")
.when(col("transaction_amount") < 10000, "high")
.otherwise("enterprise")
)
transactions_df.write \
.mode("overwrite") \
.partitionBy("tx_date", "value_tier") \
.parquet("s3://data-lake/processed/transactions/")
This produces a directory structure like:
s3://data-lake/processed/transactions/
tx_date=2024-03-22/
value_tier=low/part-00000.parquet
value_tier=medium/part-00000.parquet
value_tier=high/part-00000.parquet
value_tier=enterprise/part-00000.parquet
tx_date=2024-03-23/
value_tier=low/part-00000.parquet
...
A query for WHERE tx_date = '2024-03-22' AND value_tier = 'high' reads exactly one file. The scan reduction is multiplicative: first by date, then by tier.
Warning: With composite
partitionBy(), watch the cardinality multiplication. If you partition bydate × region × value_tier, and you have 365 days × 50 regions × 5 tiers, you have 91,250 potential directories. Even sparsely populated, that's a lot of small files and a lot of directory listing overhead. Generally, limitpartitionBy()columns to 2–3, and use in-file sorting or bucketing for additional dimensions.
You're building the output stage of a pipeline that processes a global retail platform's order data. The downstream consumers are:
customer_id for the past 90 daysrevenue_bandYour source data has these columns: order_id, customer_id, region, order_amount, order_status, created_at, updated_at.
Your task: Design and implement a composite partitioning strategy that serves all three consumers efficiently.
Before writing a line of code, document the filters each consumer uses:
| Consumer | Primary Filter | Secondary Filter | Join Key |
|---|---|---|---|
| BI Dashboard | created_at (last 7 days) |
region |
None |
| ML Pipeline | created_at (last 90 days) |
None | customer_id |
| Finance Report | created_at (monthly) |
revenue_band |
None |
All three consumers filter by date first. That makes order_date the clear outer partition key.
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
col, to_date, when, year, month
)
spark = SparkSession.builder \
.appName("orders-partition-exercise") \
.config("spark.sql.sources.partitionOverwriteMode", "dynamic") \
.getOrCreate()
# Load from your pipeline's staging area
orders_raw = spark.read.parquet("s3://data-lake/staging/orders/")
orders_enriched = orders_raw \
.withColumn("order_date", to_date(col("created_at"))) \
.withColumn(
"revenue_band",
when(col("order_amount") < 50, "micro")
.when(col("order_amount") < 200, "small")
.when(col("order_amount") < 1000, "medium")
.when(col("order_amount") < 5000, "large")
.otherwise("enterprise")
)
# The ML pipeline needs hash co-location on customer_id within date windows
# The BI dashboard needs region filtering — add region as a second partition level
# The finance report needs revenue_band — add as a third partition level
# BUT: date × region × revenue_band = too many small partitions
# Decision: partition by date and region (serves BI + finance with acceptable
# scan overhead), hash on customer_id within partitions (serves ML joins)
BUCKETS_PER_PARTITION = 8 # tune based on rows per date/region combination
orders_enriched \
.repartition(BUCKETS_PER_PARTITION, col("order_date"), col("region"), col("customer_id")) \
.write \
.mode("overwrite") \
.partitionBy("order_date", "region") \
.parquet("s3://data-lake/processed/orders_final/")
output = spark.read.parquet("s3://data-lake/processed/orders_final/")
# Check rows per partition to detect skew
output.groupBy("order_date", "region") \
.count() \
.orderBy(col("count").desc()) \
.show(20)
# Check that revenue_band distribution is reasonable within a sample partition
output.filter(
(col("order_date") == "2024-03-22") & (col("region") == "EMEA")
) \
.groupBy("revenue_band") \
.count() \
.show()
In Spark, you can confirm pruning is occurring by inspecting the query plan:
# Look for "PartitionFilters" in the physical plan
output.filter(
(col("order_date") == "2024-03-22") & (col("region") == "EMEA")
).explain("formatted")
In the output, look for lines like:
PartitionFilters: [isnotnull(order_date#12), (order_date#12 = 2024-03-22),
isnotnull(region#14), (region#14 = EMEA)]
If you see PartitionFilters in the plan, the engine is pruning. If you only see PushedFilters, the filter is being applied inside files, not at the directory level — which means your partition columns weren't properly registered.
Partitioning on customer_id directly with partitionBy("customer_id") when you have 10 million customers creates 10 million directories. Directory listing alone will time out. Use hash partitioning (repartition(N, col("customer_id"))) for high-cardinality keys, not partitionBy.
Rule of thumb: partitionBy columns should have cardinality under ~10,000 distinct values. For anything higher, use bucketing or hash repartitioning instead.
When you append new date partitions to an existing table managed by Glue or Hive Metastore, new partitions don't appear automatically. Queries return zero rows for the new date. Fix with:
-- In Athena or Presto
MSCK REPAIR TABLE orders;
-- Or more targeted:
ALTER TABLE orders ADD IF NOT EXISTS PARTITION (order_date='2024-03-23');
For automated pipelines, the fix is to add a post-write step that calls the metastore API to register new partitions.
If you write a daily batch with mode("overwrite") and default overwrite settings, Spark will delete the entire dataset before writing the new partition. Three months of history, gone.
Always set spark.sql.sources.partitionOverwriteMode to dynamic before writing partial updates to a date-partitioned table.
Hash partitioning only eliminates join shuffles when both tables use exactly the same number of buckets and the same partitioning expression. If orders uses 64 buckets on customer_id and customers uses 128 buckets on customer_id, Spark still has to shuffle. Pick a standard bucket count and document it as a pipeline contract.
Null values in a partition column create a special __HIVE_DEFAULT_PARTITION__ directory. Some query engines handle this gracefully; others struggle with it. Worse, null-partitioned rows often represent data quality problems upstream. Always coalesce or filter nulls in partition key columns before writing:
orders_enriched = orders_enriched.fillna({"region": "UNKNOWN"})
orders_enriched = orders_enriched.filter(col("order_date").isNotNull())
If your query is scanning all partitions despite a date filter, check these in order:
SHOW PARTITIONS table_name — if no partitions appear, run MSCK REPAIR TABLE.created_at (the raw timestamp) won't prune order_date (the derived date) partitions. They must match exactly.WHERE DATE(created_at) = '2024-03-22' is a function call, not a column reference — it won't prune order_date partitions even if they're the same underlying value.Partitioning is one of the highest-leverage decisions in a data pipeline. The choice you make when writing output shapes every query that will ever run against that data, often without the reader knowing why some queries are fast and others aren't.
Here's the mental model to carry forward:
The correct strategy is always derived from the query patterns of your downstream consumers. Before writing a partition scheme, interview the teams who will query the data. Sketch out their WHERE clauses. Work backward from those filters to your partition keys.
Where to go next:
The partitioning decision is never final. As your data volume grows and query patterns evolve, you'll repartition. Build your pipelines so that rewriting the output layer is cheap — keep your transformation logic separate from your write logic, and treat the partition scheme as a configuration, not a constant.
Learning Path: Data Pipeline Fundamentals