Spark SQL查询Iceberg表过慢及分区数量异常问题咨询

场景信息 #

• Iceberg表详情：7600行、115列，按time列小时分区（共28个逻辑分区），累计948个快照
• 执行的Spark代码：

Long start = System.nanoTime();
SparkSession spark = getSparkSession();
System.out.println("Session creation took (ms): " + (System.nanoTime() - start) / 1000000);
        
Dataset<Row> data = spark.sql("SELECT * FROM myschema.mytable WHERE time BETWEEN CAST('2024-07-10 14:00:00.0' AS TIMESTAMP) AND CAST('2024-07-10 15:40:00.0' AS TIMESTAMP)");
        
System.out.println("Count: " + data.count());
System.out.println("Partitions: " + data.rdd().getNumPartitions());
System.out.println("Execution took (ms): " + (System.nanoTime() - start) / 1000000);

spark.stop();

• 运行输出：

Session creation took (ms): 4333
Count: 0
Partitions: 143
Execution took (ms): 107029

• 补充情况：向表中追加数据后，同一查询仍返回0行，但查询结果的分区数持续上升；改为全表查询SELECT * FROM myschema.mytable后，执行速度提升10-12倍；相同查询用Trino执行仅需1-2秒。

核心问题解答 #

1. 为何简单查询耗时近100秒，且结果行数不影响耗时？ #

主要原因在于Spark对Iceberg元数据的处理机制和Trino存在差异，加上快照数量过多的影响：

• 你的表有948个快照，Spark默认会遍历所有快照的元数据，即使查询指定了时间范围，默认配置下的Spark Iceberg连接器可能没有高效利用Iceberg的分区 pruning 和快照过滤能力，需要逐个校验每个快照下的分区是否符合时间条件，这一步的元数据IO和处理耗时极高。
• Trino对Iceberg的元数据优化更彻底，会直接读取Iceberg的manifest文件和分区统计信息，快速跳过无关的快照和分区，所以耗时仅1-2秒。
• 全表查询时，Spark不需要对每个快照的分区做时间范围校验，直接读取最新快照的所有数据，因此执行路径更短，耗时大幅降低。

可以通过以下配置优化：

• 开启分区统计自动收集：spark.sql.iceberg.statistics-auto-collect.enabled=true
• 指定只读取最新快照：设置spark.sql.iceberg.read.latest-snapshot=true，或通过spark.sql.iceberg.read.snapshot-id指定具体快照ID
• 手动执行表优化：CALL sys.optimize_table('myschema.mytable')，生成分区统计并清理无效元数据

2. 查询结果的分区数指什么？为何持续增长且与表的物理分区数不符？ #

这里的data.rdd().getNumPartitions()是Spark RDD的并行任务数，对应查询过程中需要处理的数据文件数量，和Iceberg表的逻辑分区不是同一个概念：

• Iceberg每次写入数据都会生成新快照，每个快照下会产生若干数据文件。即使新数据不在你的查询时间范围内，Spark在元数据扫描阶段仍会为所有快照中符合初步过滤条件的文件创建RDD分区（后续才会过滤掉无数据的分区），所以随着快照数量增加，分区数会持续上升。
• 表的28个分区是逻辑上的小时分区，但每个逻辑分区下可能有多个快照生成的文件，加上948个快照的累积，最终统计的RDD分区数（143）是所有快照中匹配元数据过滤规则的文件总数，因此和逻辑分区数存在差异。

解决方法是清理旧快照并优化表：

• 过期旧快照：CALL sys.expire_snapshots('myschema.mytable', TIMESTAMP '2024-07-01 00:00:00')，保留最近的必要快照即可
• 定期合并小文件：执行CALL sys.optimize_table('myschema.mytable')，减少数据文件数量，从而降低RDD分区数

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