单个小型Parquet文件的分区策略与读取性能优化咨询

我有一个大小在5-100MB之间的单个Parquet文件。尝试按日期列创建分区时会生成多个小文件，导致读取性能下降，因此我使用repartition(1)将其合并为单个文件。目前使用Spark SQL对该文件执行类似ValuationDate BETWEEN x AND y的日期范围查询，相关代码如下：

public DataFilter applyValuationDateRangeFilter() {
    AlcyoneDate startDate = calculationContext.getDateSequence().getStartDate();
    AlcyoneDate endDate = calculationContext.getDateSequence().getEndDate();
    filterQueries.append(" AND ");
    filterQueries.append("ValuationDate BETWEEN '").append(startDate).append("' AND '").append(endDate).append("'");
    return this;
}

我尝试按ValuationDate列分区并结合repartition(1)，但未获得明显性能提升。请问如何提升读取性能？

此外，我在文件加载阶段应用过滤，代码如下：

public Dataset<Row> getDatasetForInputFileWithFiltering(String parquetFilePath, String filterQuery) {
    return getSparkSession().read().format("parquet")
            .option("inferSchema", "true")
            .option("header", "true")
            .load(parquetFilePath)
            .filter(filterQuery);
}

还有，我需要迭代所有行来填充数据，代码如下：

Iterator<Row> rowIterator = rows.toLocalIterator();
while (rowIterator.hasNext()) {
    Row row = rowIterator.next();
    // 数据填充逻辑
}

由于会出现OOM，我无法使用rows.collectAsList()；又因下游代码未处理多线程，也无法使用rows.foreach和rows.foreachPartition。

1. 给ValuationDate列添加Parquet范围索引 #

Parquet支持范围索引，写入文件时开启索引后，Spark查询日期范围时可以直接跳过不满足条件的数据块，大幅减少IO开销。
写入时的配置示例：

df.write()
  .format("parquet")
  .option("parquet.enable.indexes", "true") // 开启Parquet索引
  .option("parquet.index.type", "range") // 针对日期列使用范围索引
  .option("parquet.enable.dictionary", "true") // 开启字典编码，进一步优化列存效率
  .save("path/to/optimized-parquet");

2. 关闭Schema推断，提前定义Schema #

当前加载时使用inferSchema=true会让Spark先扫描整个文件推断Schema，增加额外的IO和时间。提前定义好Schema并直接指定，同时去掉多余的header=true（Parquet是列存格式，不需要该参数）：

import org.apache.spark.sql.types.*;

// 提前定义表结构
StructType schema = new StructType()
  .add("ValuationDate", DateType)
  .add("Column1", StringType)
  .add("Column2", DoubleType)
  // 补充其他列的类型定义
  ;

Dataset<Row> df = getSparkSession().read()
  .format("parquet")
  .schema(schema) // 直接指定Schema
  .load(parquetFilePath)
  .filter(filterQuery);

3. 调整分区粒度，避免强制合并为单个文件 #

单个100MB的Parquet文件不算过大，但分区粒度太细会导致小文件问题。建议按年月（yyyy-MM）分区，既保证每个分区文件大小合理，又能让Spark在查询时直接跳过无关分区目录：
写入时的分区示例：

import static org.apache.spark.sql.functions.*;

df.withColumn("ValuationMonth", date_format(col("ValuationDate"), "yyyy-MM"))
  .write()
  .partitionBy("ValuationMonth")
  .option("maxRecordsPerFile", 100000) // 控制每个文件的记录数，避免小文件
  .format("parquet")
  .save("path/to/partitioned-data");

查询时先过滤ValuationMonth，再过滤具体日期，Spark会先定位到目标分区，再结合索引过滤数据。

4. 优化过滤条件写法，确保谓词下推生效 #

不要用字符串拼接生成过滤条件，改用Spark列表达式API，这样Spark能更好地解析条件并执行谓词下推（Predicate Pushdown），在数据源层就完成过滤：

import static org.apache.spark.sql.functions.*;

// 替换字符串拼接的过滤逻辑
AlcyoneDate startDate = calculationContext.getDateSequence().getStartDate();
AlcyoneDate endDate = calculationContext.getDateSequence().getEndDate();
Dataset<Row> filteredDf = df.filter(col("ValuationDate").between(startDate.toSqlDate(), endDate.toSqlDate()));

同时检查Spark配置spark.sql.parquet.filterPushdown是否为true（默认开启），确保过滤逻辑下推到Parquet读取阶段。

5. 优化迭代逻辑，避免OOM #

• 若使用toLocalIterator()，需注意在迭代过程中及时释放无用对象引用，避免内存堆积；同时可适当调大Driver内存（配置spark.driver.memory），缓解OOM问题。
• 关于foreachPartition的使用：其实每个分区的迭代是单线程执行的，只要下游逻辑在单线程内可正常运行，完全可以使用该方法，不会有并发问题：

rows.foreachPartition(iterator -> {
    // 该逻辑在Executor端单线程执行，每个分区对应一个线程，无多线程冲突
    while (iterator.hasNext()) {
        Row row = iterator.next();
        // 调用下游数据填充逻辑
    }
});

内容的提问来源于stack exchange，提问作者Rohan Gala