如何通过代码查询Spark job读取数据量，统计作业层级每日数据处理量

以下方案均支持按作业维度聚合，可直接满足每日总数据处理规模统计需求：

方案1：作业运行时内嵌代码统计 #

直接在作业逻辑中加入统计代码，将结果写入日志或自定义指标后按天聚合即可：

• 针对文件类数据源（HDFS、S3、OSS等）：
  遍历DataFrame关联的输入文件，求和得到精确的输入总大小，Scala示例代码如下：

  import org.apache.hadoop.fs.Path
  
  val df = spark.read.parquet("hdfs://your/data/path")
  val hadoopConf = spark.sparkContext.hadoopConfiguration
  // 单位：字节
  val totalInputBytes = df.inputFiles.map { file =>
    val path = new Path(file)
    path.getFileSystem(hadoopConf).getFileStatus(path).getLen
  }.sum
  // 可将totalInputBytes写入业务日志或监控系统，后续按天拉取聚合
  

  PySpark示例代码如下：

  from pyspark.sql import SparkSession
  from pyarrow.fs import HadoopFileSystem
  
  spark = SparkSession.builder.appName("data_size_count").getOrCreate()
  df = spark.read.parquet("hdfs://your/data/path")
  hdfs_fs = HadoopFileSystem(host="your-namenode-host", port=9000)
  total_input_bytes = 0
  for file_path in df.inputFiles():
      # 去掉hdfs协议头适配文件系统查询
      clean_path = file_path.replace("hdfs://", "")
      total_input_bytes += hdfs_fs.get_file_info(clean_path).size
  

• 针对非文件类数据源（Kafka、JDBC、HBase等）：
  直接调用Spark优化器的统计指标获取预估输入大小，在元数据完整的场景下误差低于5%：

  // 单位：字节，为Spark Catalyst优化器预估的实际读取数据大小
  val estimatedInputBytes = df.queryExecution.optimizedPlan.stats.sizeInBytes
  

  需要完全精确的场景下，可自行根据数据源特性计算：比如Kafka统计消费的offset差值乘以平均消息大小，JDBC数据源提前执行count查询结合单行平均大小计算。

方案2：运行结束后离线拉取统计 #

不需要修改现有作业代码，直接从Spark History Server的接口批量拉取统计：

• 首先调用历史服务接口拉取当日所有结束的Spark作业ID，接口路径为http://<spark-history-host>:18080/api/v1/applications?status=completed&minDate=<当日日期>
• 遍历每个作业ID，调用stage查询接口拉取所有stage的输入字节数求和，即为该作业的总输入体量，示例命令如下：

  # 替换为实际的历史服务地址和作业ID，返回结果为作业总输入字节数
  curl http://spark-history:18080/api/v1/applications/application_1690000000000_12345/stages | jq '[.[].inputBytes] | add'
  

• 每日定时执行上述逻辑即可自动汇总得到当日所有作业的总数据处理规模。

注意事项 #

• 如果作业使用了分区裁剪、行过滤器，inputFiles方法会返回所有匹配路径的文件大小，包含未实际读取的分区，该场景优先使用Spark内部的stats指标，会自动计算实际读取的数据量
• 建议统计时将结果转换为统一单位（GB/TB），方便跨作业对比汇总

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