Spark读取Delta Lake过滤后分区数随Databricks Runtime变化问题

数据存储结构 #

我的Delta Lake存储的Parquet文件结构如下：

data/
├─ _delta_log/
├─ Year=2018/
│  ├─ Month=1/
│  │  ├─ Day=1/
│  │  │  ├─ part-01797-bd37cd91-cea6-421c-b9bb-0578796bc909.c000.snappy.parquet
│  │  ├─ ...
│  │  ├─ Day=31/
│  │  │  ├─ part-01269-3b9691cf-311a-4d26-8348-c0ced0e06bf0.c000.snappy.parquet
│  ├─ Month=2/
├─ Year=2019/

2018年1月对应31个Parquet文件（每天1个）。

测试代码与结果 #

执行以下Spark代码读取数据、过滤指定月份并查看RDD分区数：

df=spark.read.format('delta').load("/data")
df=df.filter((F.col("Year")==2018) & (F.col("Month")==1))
df.rdd.getNumPartitions()

不同Databricks Runtime（DBR）版本得到的结果不同：

• DBR 10.4：15个分区
• DBR 12.2：10个分区

问题 #

1. 为何两种情况下过滤后分区数都从31减少到15或10？
2. 为何DBR 10.4与DBR 12.2的结果有差异？由于DataFrame中存在TimeStamp列，这是否与DBR 11.2中推出的Ingestion Time Clustering有关？

解答 #

问题1：过滤后分区数减少的原因 #

Spark读取Delta Lake数据时会自动执行小文件合并优化：

• 单天对应的Parquet文件属于小文件范畴，Spark为避免小文件过多带来的任务调度开销，会根据集群资源（Executor内存、CPU核数等）自动将多个小文件合并到同一个RDD分区中，最终分区数远小于原始文件总数31。

问题2：不同DBR版本分区数差异的原因 #

DBR 10.4与12.2的分区数差异源于Spark和Delta Lake的优化策略迭代，和Ingestion Time Clustering无直接关系：

1. Spark版本迭代：DBR 10.4基于Spark 3.2，DBR 12.2基于Spark 3.3，Spark 3.3对小文件合并的阈值、算法做了优化，合并后的分区数进一步减少。
2. Delta Lake版本升级：不同DBR版本配套的Delta Lake版本不同，新版本在数据读取的文件扫描、分区规划逻辑上有优化，合并小文件的效率更高。
3. Ingestion Time Clustering是针对数据写入阶段的聚类优化，作用是减少后续查询的扫描范围，和读取时的分区合并逻辑无关，不是导致该差异的原因。

内容的提问来源于stack exchange，提问作者Oliver Angelil