📚 课程目录第3章数据采集与埋点

3、数据采集与埋点：集群资源指标采集（YARN/HDFS/Spark）、任务级粒度埋点、元数据关联

数据采集是整个成本归因的「眼睛」。眼睛瞎了，后面算得再花哨也没用。

我见过不少团队，一上来就搞复杂的成本分摊模型，结果发现底层数据全是毛刺——采集周期不对、粒度太粗、元数据对不上。说白了，地基没打好，楼盖得再高也得塌。

这一章，咱们就聊聊怎么把这双「眼睛」擦亮。

3.1 集群资源指标采集：YARN / HDFS / Spark

集群层面的指标，是成本归因的「原材料」。我习惯把它们分成三类：计算、存储、执行。

维度	核心组件	关键指标	采集方式
计算	YARN	vcore·秒、内存·GB·秒、Container 数、队列利用率	ResourceManager REST API + TimelineServer
存储	HDFS	文件大小、副本因子、Block 数、冷热数据占比	NameNode JMX + fsimage 解析
执行	Spark	Shuffle 读写量、Task 耗时、GC 时间、Stage 数	Spark History Server + Event Log

YARN 采集，我建议走两条线：

实时线：通过 ResourceManager 的 REST API（/ws/v1/cluster/apps），每 30 秒拉一次运行中的 Application 列表。注意，这个接口返回的是快照，不是增量。你需要自己维护一个状态机，判断 App 是「新提交」「运行中」还是「已结束」。
离线线：从 TimelineServer 拉历史 App 的完整指标。YARN 默认只保留 7 天，记得调大 yarn.timeline-service.ttl-ms。

我曾经踩过一个坑：YARN 的 memory-seconds 和 vcore-seconds 是 App 维度汇总值，但如果你用 Container 粒度去加总，会发现对不上。原因是 AM Container 和日志聚合 Container 也被算进去了。所以，做成本归因时，建议直接用 App 维度的汇总值，别自己拼。

HDFS 采集，核心是拿到每个目录、每个文件的真实物理存储量。NameNode 的 JMX 接口可以暴露 FsNamesystem 的统计信息，但那是集群级别的。要拿到目录级数据，我推荐两种方式：

fsimage 解析：每天凌晨跑一次 hdfs dfsadmin -fetchImage，然后用工具（比如 fsimage_reader）解析成 CSV。这里面有每个文件的路径、大小、副本数、修改时间。
HDFS 配额：对关键目录设置 hdfs dfsadmin -setSpaceQuota，通过 hdfs dfsadmin -report 获取已用空间。但注意，配额是目录级，粒度不够细。

Spark 采集，这是最容易被忽视的。很多人只盯着 YARN 的资源消耗，但 Spark 内部的 Shuffle 和 GC 才是真正的「成本黑洞」。我习惯从 Spark Event Log 里解析出每个 Task 的指标：

// 从 event log 中提取关键字段
case class SparkTaskMetrics(
  appId: String,
  stageId: Int,
  taskId: Long,
  executorRunTime: Long,   // ms
  shuffleReadBytes: Long,
  shuffleWriteBytes: Long,
  gcTime: Long,
  memoryBytesSpilled: Long,
  diskBytesSpilled: Long
)

你想想看，一个 Task 如果 Shuffle 写了几百 GB，哪怕它只用了 1 个 vcore，背后的磁盘 I/O 和网络开销也是实打实的成本。不采集这些，成本归因就是瘸腿的。

3.2 任务级粒度埋点

集群指标是「宏观」的，任务级埋点是「微观」的。两者结合，才能定位到具体是哪行代码烧了钱。

我个人习惯在三个位置埋点：

任务提交入口：记录谁提交的、什么时间、用什么队列、参数是什么。
任务执行过程：记录每个 Stage 的起止时间、输入输出量、失败重试次数。
任务结束回调：记录最终状态、总耗时、资源消耗汇总。

一个小技巧：在 Spark 的 SparkListener 里，重写 onTaskEnd 方法，把每个 Task 的 metrics 写入 Kafka。这样既能实时监控，又能离线做成本分析。我之前的项目就是这么干的，每天处理几百万条 Task 事件，延迟在秒级。

埋点的数据格式，我建议统一成 JSON，包含以下字段：

{
  "event_type": "task_end",
  "app_id": "application_1680000000000_1234",
  "task_id": 5678,
  "stage_id": 3,
  "executor_id": "1",
  "host": "node-01.example.com",
  "start_time": 1680000000000,
  "end_time": 1680000012000,
  "duration_ms": 12000,
  "metrics": {
    "executor_run_time_ms": 11500,
    "shuffle_read_bytes": 1048576,
    "shuffle_write_bytes": 2097152,
    "gc_time_ms": 800,
    "memory_spilled_bytes": 0,
    "disk_spilled_bytes": 524288
  }
}

嗯，这里要注意：不要埋得太细。我见过有人把每个 RDD 的 lineage 都埋进去，结果一天产生几个 TB 的数据，存储成本比分析收益还高。我的原则是：埋到 Task 粒度就够了，再往下就是 profiling 的范畴，不适合做长期成本归因。

3.3 元数据关联

数据采回来了，但它们是「孤岛」。YARN 的 App ID 和 HDFS 的路径怎么对应？Spark 的 Task 和 Hive 的表怎么关联？

这就是元数据关联要解决的问题。说白了，就是给每个资源消耗打上「业务标签」。

我常用的关联链路有三条：

App ID → 任务名称 → 业务线：通过 YARN 的 application_name 或 application_tags，映射到 CMDB 中的任务注册信息。
HDFS 路径 → 表名 → 数据库：通过 Hive Metastore 的 DBS 和 TBLS 表，把文件路径反向解析成表名。
Spark SQL 执行计划 → 输入/输出表：从 Spark 的 SQLExecution 事件中，提取 input_files 和 output_files，再关联到 Hive 表。

  核心思路：元数据关联不是一次性的，而是持续维护的「映射表」。我建议每天凌晨跑一次全量关联，把当天的 App、文件、表、业务线对应关系写入关系型数据库，供成本归因模型查询。

举个例子，一条完整的关联记录长这样：

app_id	task_name	hdfs_path	hive_db	hive_table	business_line	owner
app_1680000000000_1234	etl_dwd_user_daily	/user/hive/warehouse/dwd.db/user_daily	dwd	user_daily	用户增长	zhangsan

有了这张表，你就能回答「用户增长团队昨天在 HDFS 上花了多少钱」「张三提交的任务消耗了多少 vcore·秒」这类问题了。

我曾经遇到过一个情况：某个业务线的任务名全是 spark-submit-xxx 这种默认值，根本没法关联。后来我强制要求所有任务在提交时带上 --name 参数，并且和 CMDB 中的任务名保持一致。说白了，规范比技术更重要。没有规范，再好的关联算法也救不了。

3.4 整体采集架构

把上面三块串起来，就是一套完整的数据采集架构。我画了一张图，帮你理清脉络：

这张图里，我刻意把「采集层」和「存储层」分开。为什么？因为采集逻辑和存储逻辑的变更频率不一样。采集层经常要适配新的数据源（比如新增 Flink 任务），而存储层相对稳定。解耦之后，改采集不会影响下游的存储和查询。

我个人习惯：采集层用 Python 写，轻量、好维护；存储层用 Java/Scala 写，性能高、能扛大流量。中间通过 Kafka 解耦，采集层只管往 Kafka 里塞数据，存储层只管从 Kafka 里消费。这样哪边出了问题都不会互相拖累。