Spark Telemetry Listener¶

透明的大数据可观测性方案，通过 OpenTelemetry 协议采集 Spark / MapReduce 任务指标，经 Kafka 持久化到 MySQL / ClickHouse，最终通过 Grafana 可视化。

系统架构¶

graph TD
    subgraph 采集层
        SP["Spark Plugin<br/>(Task/Stage/JVM Metrics)"]
        MR["MR Collector<br/>(History Server)"]
        MA["MR Agent<br/>(字节码增强)"]
    end

    SP -- OTLP gRPC --> OTel
    MR -- OTLP gRPC --> OTel
    MA -- OTLP gRPC --> OTel

    OTel["OTel Collector"]
    Flink["Flink Metrics Consumer"]
    DB[("MySQL / ClickHouse<br/>+ Grafana 可视化")]

    OTel -- "Kafka (OTLP Protobuf)" --> Flink
    Flink --> DB

核心组件¶

支持版本¶

快速开始¶

详细的部署指南见 快速开始，包含完整的构建、基础设施部署、组件配置和数据验证步骤。

构建¶

# Omnipackage（推荐，单 JAR 支持 Spark 2/3/4 + MR Agent/Collector + Hive Hook）
chmod +x build-omni.sh && ./build-omni.sh

# 或单独构建各版本
mvn clean package -DskipTests              # Spark 3.x（默认）
mvn clean package -Pspark-2 -DskipTests    # Spark 2.x
mvn clean package -Pspark-4 -DskipTests    # Spark 4.x

部署¶

# 安装 Omnipackage 到 Spark / Hive / MR
./deploy/install-omni.sh \
  --spark-home=/opt/spark --hive-home=/opt/hive --hadoop-home=/opt/hadoop \
  --otel-endpoint=http://otel-collector:4317 -y

# 导入 Grafana 面板
./deploy/deploy-grafana.sh \
  --grafana-url=http://grafana:3000 --user=admin --password=admin

模块结构¶

spark/spark-telemetry-common/             # 核心库：配置、模型、OTel SDK、生命周期管理
spark/spark-telemetry-adapter-spark2/     # Scala 2.11 适配层，Spark 2.4
spark/spark-telemetry-adapter-spark3/     # Scala 2.12 适配层，Spark 3.5
spark/spark-telemetry-adapter-spark30/    # Scala 2.12 适配层，Spark 3.0
spark/spark-telemetry-adapter-spark32/    # Scala 2.12 适配层，Spark 3.2
spark/spark-telemetry-adapter-spark4/     # Scala 2.13 适配层，Spark 4.0
spark/spark-telemetry-dist-spark{2,3,4}/  # 各版本 Shaded Fat JAR
spark/spark-telemetry-omni-facade/        # Omnipackage Java 门面（自动检测 Spark 版本）
spark/spark-telemetry-adapters-relocated/ # 适配器重定位（v2/v3/v4 包隔离）
spark/spark-telemetry-dist-omni/          # 统一 Shaded Fat JAR（Spark 2/3/4 + MR + Hive）
mapreduce-collector/mr-telemetry-collector/             # MR 作业指标采集器（独立 Java 应用）
mapreduce-agent/mr-telemetry-agent/                 # MR 任务级 Agent（Java Agent）
mapreduce-collector/mr-telemetry-dist/         # Shaded Fat JAR
mapreduce-agent/mr-telemetry-agent-dist/         # Shaded Fat JAR
hive/hive-telemetry-hook/                # Hive 查询 Hook（ExecuteWithHookContext）
hive/hive-telemetry-hook-dist/           # Shaded Fat JAR
flink/metrics-flink-consumer/             # Flink 消费者（Kafka → MySQL / ClickHouse）
flink/metrics-flink-consumer-dist/        # Shaded Fat JAR
diagnostic/diagnostic-core/               # 交互式诊断工具（JLine CLI + 11 状态自动检查）
integration-tests/                  # 集成测试（Spark 3）

关键特性¶

• DELTA Temporality：所有 OTLP 导出器使用 DELTA 临时性，防止重导出时数据重复
• 异步刷新：flushAsync() 在 onJobEnd 时非阻塞刷新，避免阻塞 DAGScheduler
• appId Fallback：自动回退 appId → appName → "unknown"，兼容 local 模式
• 三层配置合并：Spark Conf 覆盖 > HOCON 文件 > 内置默认值
• 指标分类开关：6 个 Category 独立控制采集粒度（含 SQL 查询执行指标和 query_text 追踪）
• Stage 治理预聚合：Flink Consumer 自动计算数据倾斜、CPU 效率、GC 开销等治理指标
• SQL 文本追踪：Spark 和 Hive 均支持 query_text 捕获，自动截断到配置长度（默认 4096 字符），写入 sql_query_metrics、hive_query_metrics、metric_events 三张表
• metric_events 统一宽表：跨引擎分析大宽表，整合 Spark/MR/Hive 全部分类表数据，支持按 engine 和 event_type 维度的跨引擎聚合查询
• Shaded Fat JAR：OTel/gRPC/Protobuf 等依赖 relocate 到 x.mg.metrics.shaded.*，无依赖冲突

配置示例¶

配置文件示例见 conf/examples/ 目录：

• conf/examples/telemetry.conf.example — Spark 插件配置
• conf/examples/mr-collector.conf.example — MR Collector 配置
• conf/examples/flink-consumer.conf.example — Flink Consumer 配置
• conf/examples/hive-telemetry.conf.example — Hive Hook 配置
• diagnostic/diagnostic-core/src/main/resources/diagnostic.conf — 诊断工具配置

指标概览¶

Spark 指标¶

MR 指标¶

Hive 指标¶

Grafana 可视化¶

deploy/grafana/ 目录提供预构建仪表盘 JSON 文件，可通过 deploy/deploy-grafana.sh 一键导入：

前 6 个面板使用各引擎独立分类表，后 7 个分析面板使用 metric_events 统一宽表实现跨引擎聚合查询。覆盖：任务 IO / 时长时序趋势、JVM 内存 / GC 监控、数据倾斜检测、资源效率分析、成本归属、小文件检测、任务时长直方图分布。

完整文档¶

详细的部署指南、配置参数、指标参考、排查手册见 部署指南。

打包与发布流程见 发布指南。

K8s 测试环境¶

已迁移至裸节点测试，不再依赖 Kubernetes。详见 集成测试 README。

集成测试¶

所有集成测试在裸节点（bare-metal）上运行，通过 Docker 容器提供后端服务（OTel Collector、Kafka、MySQL），无需 K8s 集群。

测试矩阵¶

运行方式¶

# 构建并部署
mvn clean package -Pspark-3 -DskipTests
cp spark/spark-telemetry-dist-spark3/target/*.jar $SPARK_HOME/jars/

# 运行全部 IT 测试
mvn verify -Pspark-3 -pl integration-tests -Dtest.skip=true

# 运行单个测试
mvn failsafe:integration-test -Pspark-3 -pl integration-tests -Dit.test=SparkMetricsFieldVerificationIT

性能基准测试¶

使用 Intel HiBench 在 4C8G 单节点环境（<BENCHMARK_SERVER_IP>）上，对比 telemetry 组件加载前后的业务性能开销。数据规模：HiBench small profile（WordCount 320MB，SQL 100K rows，KMeans 3M samples）。每个工作负载先跑一次 baseline（无 telemetry），再跑一次 with-telemetry，验证指标到达 MySQL。

Spark 3.2.0 + Hadoop 3.2.0¶

Omnipackage 通过 spark.plugins 加载，spark.telemetry.otel.export.interval.ms=5000，开启全部指标类别。

10 个工作负载全部通过，所有指标均已验证到达 MySQL。平均开销约 -1.3%（在测量噪声范围内）。

MR Agent + Hadoop 3.2.0¶

MR Agent 通过 -javaagent 注入到 mapreduce.map/reduce.java.opts。

3 个工作负载全部通过。Telemetry 运行更快是因为 MR 任务数和 JVM 预热差异，非 agent 效果。所有 mr.task.* 指标已验证到达 MySQL。

Hive Hook + Hadoop 3.2.0¶

Hive Hook 通过 hive.exec.post.hooks 注入。测试 Hive 3.1.3 和 2.3.9，均使用 MR 引擎。

Hive 3.1.3

Hive 2.3.9

12 个 Hive 运行全部成功，所有 hive.query.* 指标已验证到达 MySQL。Hook 开销 <2%。

兼容性矩阵¶

测试环境¶

• 硬件: 4C8G 单节点（<BENCHMARK_SERVER_IP>），Java 8（/opt/jdk8u482-b08）
• 数据流: Plugin/Agent/Hook → OTLP gRPC → OTel Collector → Kafka → Flink Consumer → MySQL
• HiBench 版本: 8.0-SNAPSHOT，small profile
• Benchmark 脚本: benchmark/auto_bench.sh

异常隔离测试¶

Telemetry 组件遵循 永不阻塞用户任务 原则：任何 telemetry 初始化失败、OTel 连接断开、SDK 内部异常都不应影响 Spark/Hive/MR 作业的正常执行。

防御层次¶

验证场景¶

已知日志行为¶

当 OTel Collector 不可达时，用户将看到以下日志（WARNING 级别，不阻塞任务）：

WARNING x.mg.metrics.sparktelemetry.otel.OtelRegistry: OTel Collector connection failed, metrics will not be exported: ...

OTel SDK 内部的 gRPC 重连错误日志已降级为 WARNING，不会以 ERROR 级别输出。

License¶

Private