This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c692a4c35d [spark] Expose Paimon metrics through JMX (#10131)
c692a4c35d is described below
commit c692a4c35dcaff56f5ef089d865d581a42859926
Author: hutiefang76 <[email protected]>
AuthorDate: Thu Sep 24 10:51:57 2026 +0800
[spark] Expose Paimon metrics through JMX (#10131)
---
.../paimon/spark/commands/PaimonSparkWriter.scala | 4 ++
.../paimon/spark/metric/SparkMetricGroup.scala | 71 ++++++++++++++++++++++
.../paimon/spark/metric/SparkMetricRegistry.scala | 5 +-
.../spark/metrics/source/PaimonMetricsSource.scala | 64 +++++++++++++++++++
.../apache/paimon/spark/sql/PaimonMetricTest.scala | 70 +++++++++++++++++++++
5 files changed, 212 insertions(+), 2 deletions(-)
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 76bcbf2505..132b8bfe47 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 0000000000..fb4965b22b
--- /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 9aeeed7a03..ef5b6940cb 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 0000000000..29898060c6
--- /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 3b67841892..ede81d7ffb 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) {
