Spark按location分区遇小分区生成大量attempt文件，求最优解决方案

Hey there! Let's tackle this problem you're facing with tiny partitions and those annoying attempt* files when writing your users dataset by location. You don't have to wipe out the location field to fix this—here are several practical approaches to keep your data intact while resolving the small partition issue:

1. Tune Spark's File Size Configuration #

Spark has built-in settings to control output file sizes, which directly reduces the number of tiny files (including attempt* ones from partial task writes). Set these configurations before writing your DataFrame:

# PySpark example
spark.conf.set("spark.sql.files.maxRecordsPerFile", 1000)  # Adjust based on your average record size
spark.conf.set("spark.sql.adaptive.enabled", "true")       # Enables adaptive execution to optimize partitions
spark.conf.set("spark.sql.files.openCostInBytes", "134217728")  # 128MB, helps Spark calculate optimal file sizes

Then proceed with your original write command:

usersNewDf.write.partitionBy("location") \
    .parquet("....../parquet/users.parquet")

This tells Spark to aim for files with a targeted number of records (or size), merging small chunks into larger, more manageable files.

2. Repartition Strategically Before Writing #

You can repartition your data to group records by location while controlling the overall number of partitions. This reduces the chance of tiny partitions forming, especially if many location values have very few records:

# Repartition by location with a reasonable total partition count (adjust 200 to fit your data size)
usersNewDf.repartition(200, "location") \
    .write.partitionBy("location") \
    .parquet("....../parquet/users.parquet")

Note: Use repartition if you need a full shuffle to balance data across partitions, or coalesce if you just want to reduce partition count without shuffling (but coalesce won't rebalance by location).

3. Compact Files Post-Write with Lakehouse Formats #

If you're open to using a lakehouse format like Delta Lake (built on Parquet), you can easily compact small files after writing. Delta automatically cleans up attempt* files and optimizes partition sizes:
First, write your data as Delta instead of raw Parquet:

usersNewDf.write.partitionBy("location") \
    .format("delta") \
    .save("....../delta/users.delta")

Then run the optimize command to compact small files:

OPTIMIZE delta.`....../delta/users.delta` ZORDER BY location

This merges tiny files into larger ones and improves query performance over time.

4. Group Rare Locations into a "Misc" Category #

Instead of clearing the location field, create a new partition column that groups low-frequency locations into a single "misc" partition. This keeps all original location data while reducing partition count:

from pyspark.sql.functions import when, col

# Calculate location frequencies
location_counts = usersNewDf.groupBy("location").count().collect()
# Define threshold (e.g., locations with <5 records are rare)
rare_locations = [row.location for row in location_counts if row.count < 5]

# Create grouped partition column
users_grouped = usersNewDf.withColumn(
    "location_group",
    when(col("location").isin(rare_locations), "misc").otherwise(col("location"))
)

# Write using the grouped column
users_grouped.write.partitionBy("location_group") \
    .parquet("....../parquet/users_grouped.parquet")

You can still query the original location values by filtering within the "misc" partition, so no data is lost.

Quick Note on attempt* Files #

Just to clarify: those attempt* files typically come from task retries or partial writes for tiny partitions. When a partition has only 1 record, Spark might create a small file, and if there's any task failure/retry, it leaves behind these attempt files. Reducing small partitions directly cuts down on this issue.

内容的提问来源于stack exchange，提问作者Jiew Meng