Spark流式自定义指标

Question

我正在开发一个Spark Streaming程序,它检索Kafka流,对流进行非常基本的转换,然后将数据插入到DB(如果相关则为voltdb).我正在尝试测量向DB插入行的速率.我认为指标很有用(使用JMX).但是我找不到如何向Spark添加自定义指标.我查看了Spark的源代码并找到了这个帖子,但它对我不起作用.我还在conf.metrics文件中启用了JMX接收器.什么不起作用我没有看到我的自定义指标与JConsole.

有人可以解释如何添加自定义指标(最好通过JMX)来激发流媒体？或者如何测量我的数据库(特别是VoltDB)的插入速率？我在Java 8中使用spark.

Answer 1

在挖掘源代码后,我发现如何添加自己的自定义指标.它需要3件事:

1. 创建我自己的自定义源.有点像这样
2. 在spark metrics.properties文件中启用Jmx接收器.我使用的具体行是:*.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink它为所有实例启用JmxSink
3. 在SparkEnv指标系统中注册我的自定义源.这里可以看到一个如何操作的例子- 我之前实际上看过这个链接但错过了注册部分,这使我无法在JVisualVM中实际看到我的自定义指标

我仍在苦苦思索如何实际计算插入VoltDB的次数,因为代码在执行程序上运行,但这是一个不同主题的主题:)

我希望这会有助于其他人

Answer 2

Groupon有一个名为的库spark-metrics，可让您在执行程序上使用简单的（类似于Codahale的）API，并将结果整理回驱动程序中，并自动在Spark的现有指标注册表中注册。然后，当您根据Spark docs配置指标接收器时，这些指标会与Spark的内置指标自动导出。