From e34e3ee77757005e0ea91bba938b66e3c663929b Mon Sep 17 00:00:00 2001 From: hutiefang76 <137664623+hutiefang76@users.noreply.github.com> Date: Wed, 23 Sep 2026 10:14:19 +0800 Subject: [PATCH] [spark] Expose Paimon metrics through JMX --- .../spark/commands/PaimonSparkWriter.scala | 4 ++ .../spark/metric/SparkMetricGroup.scala | 71 +++++++++++++++++++ .../spark/metric/SparkMetricRegistry.scala | 5 +- .../metrics/source/PaimonMetricsSource.scala | 64 +++++++++++++++++ .../paimon/spark/sql/PaimonMetricTest.scala | 70 ++++++++++++++++++ 5 files changed, 212 insertions(+), 2 deletions(-) create mode 100644 paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricGroup.scala create mode 100644 paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/metrics/source/PaimonMetricsSource.scala diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala index 76bcbf25054a..132b8bfe470f 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala @@ -32,6 +32,7 @@ import org.apache.paimon.io.{CompactIncrement, DataIncrement} import org.apache.paimon.manifest.FileKind import org.apache.paimon.spark.{SparkPostponeStagedCommitter, SparkRow} import org.apache.paimon.spark.catalog.functions.BucketFunction +import org.apache.paimon.spark.metric.SparkMetricRegistry import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL, ROW_KIND_COL} import org.apache.paimon.spark.sort.TableSorter import org.apache.paimon.spark.util.OptionUtils.paimonExtensionEnabled @@ -86,6 +87,8 @@ case class PaimonSparkWriter( } } + @transient private lazy val metricRegistry = SparkMetricRegistry() + val postponeBatchWriteFixedBucket: Boolean = table.bucketMode() == POSTPONE_MODE && coreOptions.postponeBatchWriteFixedBucket() @@ -471,6 +474,7 @@ case class PaimonSparkWriter( val activeWriteBuilder = Option(directPostponeWriteBuilder).getOrElse(writeBuilder) val tableCommit = activeWriteBuilder.newCommit() + tableCommit.withMetricRegistry(metricRegistry) if (operation != null) { tableCommit.withOperation(operation) } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricGroup.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricGroup.scala new file mode 100644 index 000000000000..fb4965b22b02 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricGroup.scala @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.metric + +import org.apache.paimon.metrics.{Counter, Gauge, Histogram, MetricGroupImpl} + +import com.codahale.metrics.{Gauge => CodahaleGauge, MetricRegistry => CodahaleMetricRegistry} + +import java.util.{Map => JMap} + +import scala.collection.JavaConverters._ + +/** Keeps Paimon's Spark UI metrics while publishing their current values to JMX. */ +class SparkMetricGroup( + groupName: String, + variables: JMap[String, String], + registry: CodahaleMetricRegistry) + extends MetricGroupImpl(groupName, variables) { + + override def counter(name: String): Counter = { + val metric = super.counter(name) + register(name, () => java.lang.Long.valueOf(metric.getCount)) + metric + } + + override def gauge[T](name: String, gauge: Gauge[T]): Gauge[T] = { + val metric = super.gauge(name, gauge) + if (metric != null) { + register(name, () => metric.getValue.asInstanceOf[AnyRef]) + } + metric + } + + override def histogram(name: String, windowSize: Int): Histogram = { + val metric = super.histogram(name, windowSize) + register(name, () => java.lang.Double.valueOf(metric.getStatistics.getMean)) + metric + } + + private def register(name: String, value: () => AnyRef): Unit = { + val path = (Seq(groupName) ++ variables.asScala.toSeq.sortBy(_._1).flatMap { + case (key, variable) => Seq(key, variable) + } :+ name).mkString(".") + val jmxGauge = new CodahaleGauge[AnyRef] { + override def getValue: AnyRef = value() + } + + // Keep the latest value available to periodic scrapers; repeated operations on the same + // table replace their metrics rather than increasing the registry's cardinality. + registry.synchronized { + registry.remove(path) + registry.register(path, jmxGauge) + } + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricRegistry.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricRegistry.scala index 9aeeed7a03cb..ef5b6940cb14 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricRegistry.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricRegistry.scala @@ -18,10 +18,11 @@ package org.apache.paimon.spark.metric -import org.apache.paimon.metrics.{Gauge, Metric, MetricGroup, MetricGroupImpl, MetricRegistry} +import org.apache.paimon.metrics.{Gauge, Metric, MetricGroup, MetricRegistry} import org.apache.paimon.operation.metrics.{CommitMetrics, ScanMetrics, WriterBufferMetric} import org.apache.paimon.spark._ +import org.apache.spark.metrics.source.PaimonMetricsSource import org.apache.spark.sql.connector.metric.CustomTaskMetric import java.util.{Map => JMap} @@ -35,7 +36,7 @@ case class SparkMetricRegistry() extends MetricRegistry { override def createMetricGroup( groupName: String, variables: JMap[String, String]): MetricGroup = { - val metricGroup = new MetricGroupImpl(groupName, variables) + val metricGroup = new SparkMetricGroup(groupName, variables, PaimonMetricsSource.metricRegistry) metricGroups.put(groupName, metricGroup) metricGroup } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/metrics/source/PaimonMetricsSource.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/metrics/source/PaimonMetricsSource.scala new file mode 100644 index 000000000000..29898060c6b1 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/metrics/source/PaimonMetricsSource.scala @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.metrics.source + +import com.codahale.metrics.{MetricRegistry => CodahaleMetricRegistry} +import com.codahale.metrics.jmx.JmxReporter +import org.apache.spark.SparkEnv +import org.apache.spark.metrics.MetricsSystem + +/** Spark source for Paimon metrics. The reporter observes metrics added after source creation. */ +class PaimonMetricsSource extends Source { + + override val sourceName: String = "paimon" + + override def metricRegistry: CodahaleMetricRegistry = PaimonMetricsSource.sharedRegistry +} + +object PaimonMetricsSource { + + private val sharedRegistry = new CodahaleMetricRegistry() + + private val reporter = JmxReporter.forRegistry(sharedRegistry).inDomain("paimon").build() + reporter.start() + + // Spark snapshots a source's registry when it is registered, so the reporter above also + // observes metrics created later by scans and commits. + private val source = new PaimonMetricsSource + + @volatile private var registeredSystem: MetricsSystem = _ + + def metricRegistry: CodahaleMetricRegistry = { + Option(SparkEnv.get).foreach { + env => + val system = env.metricsSystem + if (registeredSystem ne system) { + synchronized { + if (registeredSystem ne system) { + if (system.getSourcesByName(source.sourceName).isEmpty) { + system.registerSource(source) + } + registeredSystem = system + } + } + } + } + sharedRegistry + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala index 3b6784189255..ede81d7ffb7e 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala @@ -20,10 +20,12 @@ package org.apache.paimon.spark.sql import org.apache.paimon.spark.PaimonMetrics.{RESULTED_TABLE_FILES, SCANNED_SNAPSHOT_ID, SKIPPED_TABLE_FILES} import org.apache.paimon.spark.PaimonSparkTestBase +import org.apache.paimon.spark.metric.SparkMetricRegistry import org.apache.paimon.spark.read.PaimonSplitScan import org.apache.paimon.spark.util.ScanPlanHelper import org.apache.paimon.table.source.DataSplit +import org.apache.spark.metrics.source.PaimonMetricsSource import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.DataFrame import org.apache.spark.sql.PaimonUtils.createDataset @@ -34,8 +36,76 @@ import org.apache.spark.sql.execution.metric.SQLMetric import org.apache.spark.sql.paimon.Utils import org.junit.jupiter.api.Assertions +import javax.management.ObjectName + +import java.lang.management.ManagementFactory + class PaimonMetricTest extends PaimonSparkTestBase with ScanPlanHelper { + test("Paimon Metric: registered metrics are exposed through JMX") { + val tableName = "jmx-metric-test-" + System.nanoTime() + val registry = SparkMetricRegistry() + val group = registry.createTableMetricGroup("commit", tableName) + group.gauge("lastCommitDuration", () => 42L) + val counter = group.counter("recordsWritten") + counter.inc(3) + val histogram = group.histogram("commitDuration", 10) + histogram.update(7) + + val server = ManagementFactory.getPlatformMBeanServer + val names = server.queryNames(new ObjectName("paimon:*"), null) + val metricName = names.toArray.collectFirst { + case name: ObjectName + if name.toString.contains(tableName) && + name.toString.contains("lastCommitDuration") => + name + } + assert(metricName.isDefined, s"JMX metric missing for $tableName: $names") + assert(server.getAttribute(metricName.get, "Value") == 42L) + assert(group.getMetrics.get("lastCommitDuration") != null) + + val counterName = names.toArray.collectFirst { + case name: ObjectName + if name.toString.contains(tableName) && + name.toString.contains("recordsWritten") => + name + }.get + assert(server.getAttribute(counterName, "Value") == 3L) + val histogramName = names.toArray.collectFirst { + case name: ObjectName + if name.toString.contains(tableName) && + name.toString.contains("commitDuration") => + name + }.get + assert(server.getAttribute(histogramName, "Value") == 7.0) + + val metricCount = PaimonMetricsSource.metricRegistry.getMetrics.size() + val nextGroup = registry.createTableMetricGroup("commit", tableName) + nextGroup.gauge("lastCommitDuration", () => 99L) + assert(server.getAttribute(metricName.get, "Value") == 99L) + assert(PaimonMetricsSource.metricRegistry.getMetrics.size() == metricCount) + } + + test("Paimon Metric: V1 commit metrics are exposed through JMX") { + withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") { + withTable("T_V1_JMX") { + sql("CREATE TABLE T_V1_JMX (id INT)") + sql("INSERT INTO T_V1_JMX VALUES (1), (2)") + + val server = ManagementFactory.getPlatformMBeanServer + val names = server.queryNames(new ObjectName("paimon:*"), null) + val commitMetric = names.toArray.collectFirst { + case name: ObjectName + if name.toString.toLowerCase.contains("t_v1_jmx") && + name.toString.contains("lastTableFilesAdded") => + name + } + assert(commitMetric.isDefined, s"V1 commit JMX metric missing: $names") + assert(server.getAttribute(commitMetric.get, "Value").asInstanceOf[Long] > 0L) + } + } + } + test(s"Paimon Metric: scan driver metric") { // Spark support reportDriverMetrics since Spark 3.4 if (gteqSpark3_4) {