如何创建源以将指标从Spark导出到另一个接收器(Prometheus)?

问题描述

我试图从我用Scala编写的spark应用程序创建度量标准源，以将数据导出到另一个系统，而不是Prometheus.根据该网站的数据砖，我需要创建一个扩展源特征.但是，Source特质是 private [spark]特质Source ，我的源无法将其可视化.当我创建此类时，出现错误无法从此位置访问符号源.

I am trying to create a source for metrics from my spark application written in Scala to export data to another system, preferable to Prometheus. According to this site from Data bricks I need to create a source that extends the Source trait. However, the Source trait is private[spark] trait Source and my source cannot visualize it. When I create this class I get the error Symbol Source is inaccessible from this place.

package org.sense.spark.util

import org.apache.spark.metrics.source.Source
import com.codahale.metrics.{Counter, Histogram, MetricRegistry}

class MetricSource extends Source {
  override val sourceName: String = "MySource"

  override val metricRegistry: MetricRegistry = new MetricRegistry

  val FOO: Histogram = metricRegistry.histogram(MetricRegistry.name("fooHistory"))
  val FOO_COUNTER: Counter = metricRegistry.counter(MetricRegistry.name("fooCounter"))
}

如何创建将数据导出到Prometheus的源?我想从 combineByKey 转换内的UDF导出监视值.这些值将是聚合的延迟时间和此转换的吞吐量IN/OUT.

How can I create my source to export data to Prometheus? I would like to export monitored values from a UDF inside the combineByKey transformation. The values would be latency to aggregate and throughput IN/OUT of this transformation.

如果需要检查我正在使用的库，这是我的 build.sbt 文件.

This is my build.sbt file in case it is necessary to check the libraries that I am using.

name := "explore-spark"

version := "0.2"

scalaVersion := "2.12.3"

val sparkVersion = "3.0.0"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion,
  "org.apache.spark" %% "spark-streaming" % sparkVersion % "provided",
  "org.apache.spark" %% "spark-sql" % sparkVersion % "provided",
  "com.twitter" %% "algebird-core" % "0.13.7",
  "joda-time" % "joda-time" % "2.5",
  "org.fusesource.mqtt-client" % "mqtt-client" % "1.16"
)

mainClass in(Compile, packageBin) := Some("org.sense.spark.app.App")
mainClass in assembly := Some("org.sense.spark.app.App")

assemblyOption in assembly := (assemblyOption in assembly).value.copy(includeScala = false)
assemblyJarName in assembly := s"${name.value}_${scalaBinaryVersion.value}-fat_${version.value}.jar"

推荐答案

您需要将扩展​​Source的类放在与source相同的包中

You will need to put your class which extends Source in the same package as source

package org.apache.spark.metrics.source

import com.codahale.metrics.{Counter, Histogram, MetricRegistry}

class MetricsSource extends Source {
  override val sourceName: String = "MySource"

  override val metricRegistry: MetricRegistry = new MetricRegistry

    val FOO: Histogram = metricRegistry.histogram(MetricRegistry.name("fooHistory"))

  val FOO_COUNTER: Counter = metricRegistry.counter(MetricRegistry.name("fooCounter"))
}

这篇关于如何创建源以将指标从Spark导出到另一个接收器(Prometheus)?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！