如何计算Spark查询实际扫描的字节数（含Iceberg表适配方案）

核心需求 #

• 基础目标：获取查询在文件层面实际读取的字节数
• 进阶目标：量化谓词下推后，实际读取数据相对全表数据的缩减幅度
• 最终目的：从数据吞吐量维度，验证存储布局与表分区策略的有效性

通用Spark度量方案 #

1. 执行后实际读取量（最精准） #

Spark执行查询后，可通过UI或代码获取实际读取的字节数：

• Spark UI：进入对应Job的Stage详情页，查看Input Size字段，这是任务从存储实际读取的总字节数（需排除Shuffle Read数据）
• 代码层面提取：触发查询执行后，通过QueryExecution的metrics获取扫描阶段的实际读取量：

  val df = spark.sql("SELECT a,b,c FROM table WHERE c = 'foo'")
  df.collect() // 触发执行
  // 提取扫描节点的度量指标
  val scanMetrics = df.queryExecution.executedPlan.collect {
    case scan: org.apache.spark.sql.execution.ScanExec => scan.metrics
  }.head
  // 实际读取的字节数
  val actualBytesRead = scanMetrics("inputBytes").value
  

2. 预执行阶段的估算（计算缩减幅度） #

如果表已通过ANALYZE TABLE生成统计信息，可通过EXPLAIN COST获取预估读取量，结合全表数据量计算缩减比例：

• 执行估算命令：

  EXPLAIN COST SELECT a,b,c FROM table WHERE c = 'foo'
  

• 输出会包含Estimated Size字段，代表谓词过滤后预计读取的字节数
• 结合表的总字节数计算预估缩减比例：

  val totalTableBytes = spark.table("table").stats.sizeInBytes
  val estimatedReductionRatio = 1 - (estimatedFilteredBytes.toDouble / totalTableBytes)
  

Iceberg表专属方案 #

Iceberg的元数据设计（文件级统计、分区快照）提供了更精准的度量能力，适合验证分区策略有效性：

1. 预执行阶段的精准估算 #

通过Iceberg API直接查询元数据，获取全表总字节数和符合谓词的文件总字节数：

import org.apache.iceberg.expressions.Expressions
import org.apache.spark.sql.iceberg.IcebergTable

val icebergTable = spark.table("table").asInstanceOf[IcebergTable].table()
// 全表总字节数（基于当前快照）
val totalTableBytes = icebergTable.currentSnapshot().totalFileSizeInBytes()
// 构造查询谓词（c='foo'）
val predicate = Expressions.equal("c", "foo")
// 估算符合谓词的文件总字节数（基于Iceberg文件级统计）
val estimatedFilteredBytes = icebergTable.newScan().filter(predicate).totalByteCount()
// 预估缩减幅度
val estimatedReductionRatio = 1 - (estimatedFilteredBytes.toDouble / totalTableBytes)

2. 执行后实际扫描与跳过的字节数 #

Iceberg在Spark执行时会暴露专属度量指标，可直接获取实际扫描和跳过的字节数：

val df = spark.sql("SELECT a,b,c FROM table WHERE c = 'foo'")
df.collect()
// 提取Iceberg扫描节点的度量
val icebergMetrics = df.queryExecution.executedPlan.collect {
  case scan: org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExec =>
    scan.metrics.filter(_._1.startsWith("iceberg."))
}.head
// 实际扫描的字节数
val actualScannedBytes = icebergMetrics("iceberg.scanned.bytes").value
// 谓词下推跳过的字节数
val skippedBytes = icebergMetrics("iceberg.skipped.bytes").value
// 实际缩减幅度
val actualReductionRatio = skippedBytes.toDouble / totalTableBytes

3. 分区策略有效性验证 #

通过EXPLAIN EXTENDED查看Iceberg的分区过滤细节，确认哪些分区被跳过：

EXPLAIN EXTENDED SELECT a,b,c FROM table WHERE c = 'foo'

输出中会包含Iceberg分区过滤的具体信息，直观展示分区策略对数据扫描的优化效果。

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