Spark Telemetry Listener¶

A transparent big data observability solution that captures Spark / MapReduce task metrics via the OpenTelemetry protocol, persists them through Kafka to MySQL / ClickHouse, and visualizes them in Grafana.

System Architecture¶

graph TD
    subgraph Collection Layer
        SP["Spark Plugin<br/>(Task/Stage/JVM Metrics)"]
        MR["MR Collector<br/>(History Server)"]
        MA["MR Agent<br/>(Bytecode Instrumentation)"]
    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 Visualization")]

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

Core Components¶

Supported Versions¶

Quick Start¶

For a detailed deployment guide, see Quick Start, which covers end-to-end build, infrastructure deployment, component configuration, and data validation steps.

Build¶

# Omnipackage (recommended, single JAR supporting Spark 2/3/4 + MR Agent/Collector + Hive Hook)
chmod +x build-omni.sh && ./build-omni.sh

# Or build individual versions
mvn clean package -DskipTests              # Spark 3.x (default)
mvn clean package -Pspark-2 -DskipTests    # Spark 2.x
mvn clean package -Pspark-4 -DskipTests    # Spark 4.x

Deploy¶

# Install Omnipackage to 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

# Import Grafana dashboards
./deploy/deploy-grafana.sh \
  --grafana-url=http://grafana:3000 --user=admin --password=admin

Module Structure¶

spark/spark-telemetry-common/             # Core library: config, models, OTel SDK, lifecycle management
spark/spark-telemetry-adapter-spark2/     # Scala 2.11 adapter, Spark 2.4
spark/spark-telemetry-adapter-spark3/     # Scala 2.12 adapter, Spark 3.5
spark/spark-telemetry-adapter-spark30/    # Scala 2.12 adapter, Spark 3.0
spark/spark-telemetry-adapter-spark32/    # Scala 2.12 adapter, Spark 3.2
spark/spark-telemetry-adapter-spark4/     # Scala 2.13 adapter, Spark 4.0
spark/spark-telemetry-dist-spark{2,3,4}/  # Version-specific Shaded Fat JARs
spark/spark-telemetry-omni-facade/        # Omnipackage Java facade (auto-detects Spark version)
spark/spark-telemetry-adapters-relocated/ # Adapter relocation (v2/v3/v4 package isolation)
spark/spark-telemetry-dist-omni/          # Unified Shaded Fat JAR (Spark 2/3/4 + MR + Hive)
mapreduce-collector/mr-telemetry-collector/             # MR job metric collector (standalone Java app)
mapreduce-agent/mr-telemetry-agent/                 # MR task-level 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 query hook (ExecuteWithHookContext)
hive/hive-telemetry-hook-dist/           # Shaded Fat JAR
flink/metrics-flink-consumer/             # Flink consumer (Kafka -> MySQL / ClickHouse)
flink/metrics-flink-consumer-dist/        # Shaded Fat JAR
diagnostic/diagnostic-core/               # Interactive diagnostic tool (JLine CLI + 11-state auto check)
integration-tests/                  # Integration tests (Spark 3)

Key Features¶

• DELTA Temporality: All OTLP exporters use DELTA temporality to prevent duplicate data on re-export
• Async Flush: flushAsync() non-blocking flush on onJobEnd to avoid blocking DAGScheduler
• appId Fallback: Automatic fallback appId -> appName -> "unknown", compatible with local mode
• Three-Tier Config Merge: Spark Conf overrides > HOCON file > built-in defaults
• Metric Category Switches: 6 independent Categories controlling collection granularity (including SQL query execution metrics and query_text tracking)
• Stage Governance Pre-aggregation: Flink Consumer automatically computes data skew, CPU efficiency, GC overhead and other governance metrics
• SQL Text Tracking: Both Spark and Hive support query_text capture, auto-truncated to configured length (default 4096 chars), written to sql_query_metrics, hive_query_metrics, and metric_events tables
• metric_events Unified Wide Table: Cross-engine analysis wide table consolidating all Spark/MR/Hive category table data, supporting cross-engine aggregation queries by engine and event_type dimensions
• Shaded Fat JAR: OTel/gRPC/Protobuf dependencies relocated to x.mg.metrics.shaded.*, no dependency conflicts

Configuration Examples¶

Configuration file examples are in the conf/examples/ directory:

• conf/examples/telemetry.conf.example -- Spark plugin configuration
• conf/examples/mr-collector.conf.example -- MR Collector configuration
• conf/examples/flink-consumer.conf.example -- Flink Consumer configuration
• conf/examples/hive-telemetry.conf.example -- Hive Hook configuration
• diagnostic/diagnostic-core/src/main/resources/diagnostic.conf -- Diagnostic tool configuration

Metrics Overview¶

Spark Metrics¶

MR Metrics¶

Hive Metrics¶

Grafana Visualization¶

The deploy/grafana/ directory provides pre-built dashboard JSON files, importable with deploy/deploy-grafana.sh:

The first 6 dashboards use engine-specific category tables; the latter 7 analysis dashboards use the metric_events unified wide table for cross-engine aggregation queries. Coverage includes: task IO / duration time-series trends, JVM memory / GC monitoring, data skew detection, resource efficiency analysis, cost attribution, small file detection, task duration histogram distributions.

Full Documentation¶

For detailed deployment guides, configuration parameters, metric references, and troubleshooting manuals, see the Deployment Guide.

For packaging and release processes, see the Release Guide.

K8s Test Environment¶

Testing has migrated to bare-metal nodes and no longer depends on Kubernetes. See the Integration Tests README for details.

Integration Tests¶

All integration tests run on bare-metal nodes, using Docker containers for backend services (OTel Collector, Kafka, MySQL), no K8s cluster required.

Test Matrix¶

Running¶

# Build and deploy
mvn clean package -Pspark-3 -DskipTests
cp spark/spark-telemetry-dist-spark3/target/*.jar $SPARK_HOME/jars/

# Run all IT tests
mvn verify -Pspark-3 -pl integration-tests -Dtest.skip=true

# Run a single test
mvn failsafe:integration-test -Pspark-3 -pl integration-tests -Dit.test=SparkMetricsFieldVerificationIT

Performance Benchmarks¶

Using Intel HiBench on a 4C8G single-node environment (<BENCHMARK_SERVER_IP>), comparing business performance overhead before and after loading telemetry components. Data scale: HiBench small profile (WordCount 320MB, SQL 100K rows, KMeans 3M samples). Each workload runs once as baseline (no telemetry), then once with-telemetry, verifying metrics arrive in MySQL.

Spark 3.2.0 + Hadoop 3.2.0¶

Omnipackage loaded via spark.plugins, spark.telemetry.otel.export.interval.ms=5000, all metric categories enabled.

All 10 workloads passed, all metrics verified arriving in MySQL. Average overhead approximately -1.3% (within measurement noise).

MR Agent + Hadoop 3.2.0¶

MR Agent injected via -javaagent in mapreduce.map/reduce.java.opts.

All 3 workloads passed. Telemetry runs faster due to MR task count and JVM warmup variations, not agent effect. All mr.task.* metrics verified arriving in MySQL.

Hive Hook + Hadoop 3.2.0¶

Hive Hook injected via hive.exec.post.hooks. Tested with Hive 3.1.3 and 2.3.9, both using MR engine.

Hive 3.1.3

Hive 2.3.9

All 12 Hive runs succeeded, all hive.query.* metrics verified arriving in MySQL. Hook overhead <2%.

Compatibility Matrix¶

Test Environment¶

• Hardware: 4C8G single node (<BENCHMARK_SERVER_IP>), Java 8 (/opt/jdk8u482-b08)
• Data Flow: Plugin/Agent/Hook -> OTLP gRPC -> OTel Collector -> Kafka -> Flink Consumer -> MySQL
• HiBench Version: 8.0-SNAPSHOT, small profile
• Benchmark Script: benchmark/auto_bench.sh

Error Isolation Testing¶

Telemetry components follow the principle of never blocking user tasks: any telemetry initialization failure, OTel connection loss, or SDK internal exception must never affect normal execution of Spark/Hive/MR jobs.

Defense Layers¶

Verification Scenarios¶

Known Logging Behavior¶

When the OTel Collector is unreachable, users will see the following log output (WARNING level, does not block tasks):

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

OTel SDK internal gRPC reconnection error logs have been downgraded to WARNING and will not output at ERROR level.

License¶

Private