Quick Start — Spark / MR / Hive Telemetry 部署指南¶

# 前置：JDK 8, Maven 3.6+
git clone <repo-url> && cd spark-telemetry-listener

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

# 产物位置
ls spark/spark-telemetry-dist-omni/target/spark-telemetry-dist-omni-*.jar

# 单独构建各组件（可选）
mvn clean package -pl hive/hive-telemetry-hook,hive/hive-telemetry-hook-dist -am -DskipTests      # Hive Hook
mvn clean package -pl flink/metrics-flink-consumer,flink/metrics-flink-consumer-dist -am -DskipTests  # Flink Consumer

# deploy/otel-collector/config.yaml
extensions:
  health_check:
    endpoint: 0.0.0.0:13133

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317

exporters:
  kafka:
    topic: telemetry-metrics
    encoding: otlp_proto
    brokers:
      - kafka:9092

service:
  extensions: [health_check]
  pipelines:
    metrics:
      receivers: [otlp]
      exporters: [kafka]

# Docker 部署
docker run -d --name otel-collector --network host \
  -v $(pwd)/deploy/otel-collector/config.yaml:/etc/otelcol-contrib/config.yaml \
  otel/opentelemetry-collector-contrib:0.96.0 \
  --config=/etc/otelcol-contrib/config.yaml

# KRaft 模式（无需 ZooKeeper）
docker run -d --name kafka --network host \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://$(hostname):9092 \
  apache/kafka:3.7.0

# 创建 topic
docker exec kafka /opt/kafka/bin/kafka-topics.sh --create \
  --topic telemetry-metrics --bootstrap-server localhost:9092 \
  --partitions 3 --replication-factor 1

# MySQL
docker run -d --name mysql --network host \
  -e MYSQL_ROOT_PASSWORD=root123 \
  -e MYSQL_DATABASE=metrics_db \
  mysql:8.0

# ClickHouse（可选）
docker run -d --name clickhouse --network host \
  clickhouse/clickhouse-server:23.8

# 创建配置文件 flink-consumer.conf
# 参考 conf/flink/flink-consumer-mysql.conf 或 flink-consumer-clickhouse.conf

# 独立运行（不需要 Flink 集群，使用 Flink LocalEnvironment）
java -jar metrics-flink-consumer-dist-1.0.0-SNAPSHOT.jar /path/to/flink-consumer.conf

# 提交到 Flink 集群（推荐生产环境）
/opt/flink-1.18.0/bin/flink run -c x.mg.metrics.flink.Main \
  -m localhost:8081 \
  /path/to/metrics-flink-consumer-dist-1.0.0-SNAPSHOT.jar /path/to/flink-consumer.conf

# 后台提交（断开 SSH 不中断）
nohup /opt/flink-1.18.0/bin/flink run -c x.mg.metrics.flink.Main \
  -m localhost:8081 \
  /path/to/metrics-flink-consumer-dist-1.0.0-SNAPSHOT.jar /path/to/flink-consumer.conf \
  > /tmp/flink-submit.log 2>&1 &

# 查看集群作业状态
curl -s http://localhost:8081/jobs | python3 -c \
  'import json,sys; [print(j["id"],j["status"]) for j in json.load(sys.stdin)["jobs"]]'

# 取消作业
/opt/flink-1.18.0/bin/flink cancel <job-id>

docker run -d --name grafana --network host \
  -e GF_SECURITY_ADMIN_PASSWORD=admin123 \
  grafana/grafana:latest

scp spark-telemetry-dist-omni-*.jar node:/opt/spark-telemetry-plugin.jar

spark-submit --master yarn \
  --jars /opt/spark-telemetry-plugin.jar \
  --conf spark.plugins=x.mg.metrics.sparktelemetry.adapter.SparkTelemetryPlugin \
  --conf spark.telemetry.otel.exporter.endpoint=http://otel-collector:4317 \
  --conf spark.telemetry.otel.service.name=my-spark-app \
  your-app.jar

spark-submit --master yarn \
  --jars /opt/spark-telemetry-plugin.jar \
  --conf spark.extraListeners=x.mg.metrics.sparktelemetry.adapter.SparkTelemetryListener \
  --conf spark.telemetry.otel.exporter.endpoint=http://otel-collector:4317 \
  --conf spark.telemetry.otel.service.name=my-spark-app \
  your-app.jar

# 必填
spark.telemetry.otel.exporter.endpoint=http://otel-collector:4317

# 推荐配置
spark.telemetry.otel.service.name=my-spark-app          # 服务名，Grafana 过滤用
spark.telemetry.otel.export.interval.ms=10000            # 导出间隔（默认10s）
spark.telemetry.config.path=/etc/telemetry.conf          # 配置文件路径

# 指标开关（默认值如下）
spark.telemetry.metrics.task.execution=true              # task CPU/GC/内存/耗时
spark.telemetry.metrics.task.shuffle-extended=true       # shuffle 详细指标
spark.telemetry.metrics.task.info=true                   # task host/locality 信息
spark.telemetry.metrics.stage.detailed=true              # stage 级指标
spark.telemetry.metrics.job.lifecycle=true               # job 生命周期事件
spark.telemetry.metrics.sql.query-execution=true         # SQL 执行指标（join/shuffle bytes/query text）
spark.telemetry.sql.max-length=4096                      # SQL 文本最大截断长度（字符）

# 1. 将 JAR 放到 Hadoop 节点
scp spark-telemetry-dist-omni-*.jar node:/opt/hadoop/lib/mr-telemetry-agent.jar

# 2. 配置 mapred-site.xml

<!-- mapred-site.xml -->
<property>
  <name>mapreduce.framework.name</name>
  <value>yarn</value>
</property>
<property>
  <name>mapreduce.map.java.opts</name>
  <value>-javaagent:/opt/hadoop/lib/mr-telemetry-agent.jar -Dmr.telemetry.agent.otel.exporter.endpoint=http://otel-collector:4317</value>
</property>
<property>
  <name>mapreduce.reduce.java.opts</name>
  <value>-javaagent:/opt/hadoop/lib/mr-telemetry-agent.jar -Dmr.telemetry.agent.otel.exporter.endpoint=http://otel-collector:4317</value>
</property>

# 1. 创建配置文件 mr-collector.conf
# 参考 conf/examples/mr-collector.conf.example 或 conf/spark35-hadoop3-hive2.3.9/mr-collector.conf

# mr-collector.conf
mr-telemetry {
  history-server {
    url = "http://hadoop-historyserver:19888"
    poll.interval.secs = 30
  }
  otel {
    exporter.endpoint = "http://otel-collector:4317"
    service.name = "mr-telemetry-collector"
    export.interval.ms = 10000
  }
  state {
    file = "/tmp/mr-telemetry-state.json"   # 断点续传，重启后不重复采集
  }
  collection {
    job.counters = true      # job 级别：HDFS IO, CPU, GC, maps/reduces 数
    task.counters = false    # task 级别（数据量大，按需开启）
    job.details = true       # job 名称、用户、状态
  }
}

# 2. 运行（使用 Omnipackage）
java -jar spark-telemetry-dist-omni-*.jar --mr-collector /path/to/mr-collector.conf

# 3. 生产环境建议用 systemd 或 nohup 后台运行
nohup java -jar spark-telemetry-dist-omni-*.jar --mr-collector /etc/mr-collector.conf &

# 1. 将 Omnipackage JAR 放到 HiveServer2 auxlib（自动加载，无需 hive.aux.jars.path）
cp spark-telemetry-dist-omni-*.jar $HIVE_HOME/auxlib/

<!-- hive-site.xml -->
<property>
  <name>hive.exec.post.hooks</name>
  <value>x.mg.metrics.hivetelemetry.HiveTelemetryHook</value>
</property>
<property>
  <name>hive.telemetry.otel.exporter.endpoint</name>
  <value>http://otel-collector:4317</value>
</property>
<property>
  <name>hive.telemetry.sql.max-length</name>
  <value>4096</value>
</property>

docker logs otel-collector --tail=50 | grep -E "spark\.|mr\.|hive\."

docker exec kafka /opt/kafka/bin/kafka-dump-log.sh \
  --files /tmp/kafka-logs/telemetry-metrics-0/00000000000000000000.log

-- Spark 指标
SELECT COUNT(*) FROM task_metrics;
SELECT COUNT(*) FROM sql_query_metrics;

-- MR 指标
SELECT COUNT(*) FROM mr_job_metrics;    -- MR Collector
SELECT COUNT(*) FROM mr_task_metrics;   -- MR Agent

-- Hive 指标
SELECT COUNT(*) FROM hive_query_metrics;
SELECT COUNT(*) FROM hive_table_io_metrics;

-- 统一宽表
SELECT engine, event_type, COUNT(*) FROM metric_events GROUP BY engine, event_type;

# 构建诊断工具
mvn clean package -pl diagnostic/diagnostic-core -am -DskipTests

# 运行（交互式 CLI，JLine 终端）
java -jar diagnostic/diagnostic-core/target/diagnostic-core-1.0.0-SNAPSHOT.jar

# 指定配置文件
java -jar diagnostic/diagnostic-core/target/diagnostic-core-1.0.0-SNAPSHOT.jar \
  --config /path/to/diagnostic.conf

URL: mysql:3306
Database: metrics_db
User: metrics / Password: metrics