Parquet存储与Spark优化：分区过滤下推及小文件处理咨询

问题背景 #

我通过每日批处理生成仅包含单日数据的DataFrame，并将其存储至Azure Data Lake Gen 2（HDFS兼容），存储代码如下：

TradesWritePath="abfss://refined@"+datalakename+".dfs.core.windows.net/curated/clientscoredata/clientscoredatainput/Trades/" + TradeDatedt + "/Trades_" + tradedateInt + ".parquet"
tradesdf.write.mode("overwrite").parquet(TradesWritePath)

因仅存单日数据，未对DataFrame进行分区，数据按日期路径（如Trades/2019/08/25、Trades/2019/08/26）存储。现咨询以下问题：

1. 数据全部存储完成后，按日期过滤的谓词能否下推？HDFS是否会直接定位数据而非执行全量扫描？是否仍需使用Partition by写入，以让Spark和HDFS支持过滤下推？
2. 存储目录下存在大量小文件part-****.snappy.parquet，已知可通过repartition减少文件数量，但此举是否必要？若按128MB分片存储以平衡NameNode负载与文件大小，该策略是否适用于snappy压缩的Parquet分片文件？

问题解答 #

1. 谓词下推与路径过滤的有效性 #

先给你吃个定心丸：你现在按日期路径存储的方式，完全可以让Spark实现谓词下推、避免全量扫描，但前提是查询逻辑要和路径结构匹配。

举个例子，当你用类似spark.read.parquet("abfss://.../Trades/*/*/*").filter("trade_date = '2019-08-25'")的语句查询时，如果Spark能识别到路径里的日期和过滤条件的对应关系，它会直接跳过不符合的目录，只扫描目标日期的文件。这里有个关键细节：你得让Spark“看懂”你的路径结构——要么保持默认开启的spark.sql.sources.partitionColumnTypeInference.enabled=true（Spark会自动推断路径中的分区列），要么读取时显式指定基准路径，比如：

df = spark.read.option("basePath", "abfss://.../Trades/") \
              .parquet("abfss://.../Trades/*/*/*")

这样Spark会把2019/08/25解析成year=2019, month=08, day=25这类分区列，后续过滤日期时就会精准定位到对应目录，完全不会碰其他日期的数据。

那要不要换成partitionBy写入？其实两种方式本质都是基于目录的分区：partitionBy是Spark帮你自动生成规范的分区目录（比如Trades/trade_date=2019-08-25/），而你现在是手动构造路径。区别在于：

• 用partitionBy的话，Spark会自动维护分区元数据，后续读取时不用额外配置就能识别分区列，长期维护更省心；
• 手动路径的话需要确保查询过滤条件和路径结构严格匹配，否则可能触发不了下推。

如果你的路径是规范的年/月/日层级，两种方式都能实现谓词下推，不需要强制换partitionBy，但从长期维护和减少人为失误的角度，partitionBy更友好。

2. 小文件问题与128MB分片策略的适用性 #

首先明确：减少小文件非常有必要，理由如下：

• NameNode要维护每个文件的元数据，大量小文件会疯狂占用NameNode的内存，拖慢集群响应甚至影响稳定性；
• Spark读取大量小文件时，会启动超多任务，每个任务只处理一点点数据，任务调度的开销远大于数据处理本身，查询性能会断崖式下降；
• 小文件的压缩效率也更低，snappy这类算法在处理大体积数据时，压缩比会更高。

再说说128MB分片策略：这个方案完全适配snappy压缩的Parquet文件。其实Spark针对HDFS场景的默认分区大小就是128MB，这个数值是经过大量实践验证的，能很好平衡NameNode负载、任务并行度和压缩效率。

具体操作上，你可以在写入前用repartition或coalesce调整分区数：

• 如果每日数据量相对固定，直接计算分区数：总数据大小 ÷ 128MB，用repartition(计算出的分区数)即可；
• 如果数据量波动大，可以保留默认的spark.sql.files.maxPartitionBytes=134217728（即128MB），但写入时还是建议显式用repartition来精准控制。

另外，也可以在每日批处理结束后，对当日的小文件做合并操作，比如：

spark.read.parquet(当日路径).repartition(1).write.mode("overwrite").parquet(当日路径)

这样既保证了当日数据是一个（或少量）大文件，又不会影响后续的谓词下推。

内容的提问来源于stack exchange，提问作者Saugat Mukherjee