Approach: Read only the date partitions needed (partition pruning), select only required columns, use built-in map-side combining and an approximate distinct for user_count to avoid large shuffles; tune shuffle partitions. Provide exact-distinct option if strict accuracy required (but higher shuffle cost).python
from pyspark.sql import functions as F
# params
src_path = "s3://bucket/events/" # parquet partitioned by event_date=YYYY-MM-DD
dest_path = "s3://bucket/rollups/daily/"
start_date = "2025-11-01"
end_date = "2025-11-30"
# Read with partition pruning: filter on partition column
df = (spark.read
.parquet(src_path)
.select("event_date", "event_type", "user_id") # project only needed cols
.where((F.col("event_date") >= start_date) & (F.col("event_date") <= end_date))
)
# Option A: approximate distinct (fast, low shuffle)
rollup_approx = (df.groupBy("event_date", "event_type")
.agg(
F.count("*").alias("event_count"),
F.approx_count_distinct("user_id").alias("user_count") # HyperLogLog
))
# Option B: exact distinct (more expensive shuffle)
# from pyspark.sql import functions as F
# rollup_exact = (df.groupBy("event_date", "event_type")
# .agg(
# F.count("*").alias("event_count"),
# F.countDistinct("user_id").alias("user_count")
# ))
# Performance tuning: reduce unnecessary shuffles
spark.conf.set("spark.sql.shuffle.partitions", "200") # adjust to cluster size
# Persist if reused
rollup_approx = rollup_approx.repartition("event_date", "event_type") # co-locate keys for downstream writes
# Write partitioned by event_date for efficient queries later
(rollup_approx.write
.mode("overwrite")
.partitionBy("event_date")
.parquet(dest_path))
