Scala StreamingListenerBatchCompleted类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Scala›Scala教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Scala中org.apache.spark.streaming.scheduler.StreamingListenerBatchCompleted类的典型用法代码示例。如果您正苦于以下问题：Scala StreamingListenerBatchCompleted类的具体用法？Scala StreamingListenerBatchCompleted怎么用？Scala StreamingListenerBatchCompleted使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了StreamingListenerBatchCompleted类的3个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Scala代码示例。

示例1: StreamingMetricsListener

//设置package包名称以及导入依赖的类
package com.groupon.dse.spark.listeners

import org.apache.spark.groupon.metrics.UserMetricsSystem
import org.apache.spark.streaming.scheduler.{StreamingListener, StreamingListenerBatchCompleted}


class StreamingMetricsListener extends StreamingListener {
  private lazy val processingTimeHistogram = UserMetricsSystem.histogram("baryon.processingTime")
  private lazy val schedulingDelayHistogram = UserMetricsSystem.histogram("baryon.schedulingDelay")

  override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
    if (batchCompleted.batchInfo.processingDelay.isDefined) {
      processingTimeHistogram.update(batchCompleted.batchInfo.processingDelay.get)
    }
    if (batchCompleted.batchInfo.schedulingDelay.isDefined) {
      schedulingDelayHistogram.update(batchCompleted.batchInfo.schedulingDelay.get)
    }
  }
}

开发者ID:groupon，项目名称:baryon，代码行数:20，代码来源:StreamingMetricsListener.scala

示例2: PrometheusSparkMetrics

//设置package包名称以及导入依赖的类
package com.godatadriven.twitter_classifier

import io.prometheus.client.exporter.PushGateway
import io.prometheus.client.{CollectorRegistry, Gauge}
import org.apache.spark.streaming.scheduler.{StreamingListener, StreamingListenerBatchCompleted}

class PrometheusSparkMetrics(sparkJob: String) extends StreamingListener {

  override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
    val registry: CollectorRegistry = new CollectorRegistry()
    val pushGateway: PushGateway = new PushGateway("127.0.0.1:9091")
    addInputRate(batchCompleted, registry)
    addSchedulingDelay(batchCompleted, registry)
    addProcessingTime(batchCompleted, registry)
    addTotalDelay(batchCompleted, registry)
    pushGateway.push(registry, "spark_streaming_exporter")
  }

  def addInputRate(batchCompleted: StreamingListenerBatchCompleted, registry: CollectorRegistry): Unit = {
    addMetric(registry, batchCompleted.batchInfo.numRecords, "spark_streaming_input_rate", "The input rate of our spark streaming job")
  }

  def addSchedulingDelay(batchCompleted: StreamingListenerBatchCompleted, registry: CollectorRegistry) = {
    addMetric(registry, batchCompleted.batchInfo.schedulingDelay.get, "spark_streaming_scheduling_delay", "The scheduling delay of our spark streaming job")
  }
  def addProcessingTime(batchCompleted: StreamingListenerBatchCompleted, registry: CollectorRegistry) = {
    addMetric(registry, batchCompleted.batchInfo.processingDelay.get, "spark_streaming_processing_time", "The processing delay of our spark streaming job")
  }

  def addTotalDelay(batchCompleted: StreamingListenerBatchCompleted, registry: CollectorRegistry) = {
    addMetric(registry, batchCompleted.batchInfo.totalDelay.get, "spark_streaming_total_delay", "The total delay of our spark streaming job")
  }

  def addMetric(registry: CollectorRegistry, value: Double, name: String, helpText: String): Unit = {
    val totalDelay: Gauge = Gauge.build()
      .help(helpText)
      .name(name)
      .labelNames("spark_job")
      .register(registry)
    totalDelay.labels(sparkJob).set(value)
  }
}

开发者ID:rweverwijk，项目名称:twitter-to-neo4j，代码行数:43，代码来源:PrometheusSparkMetrics.scala

示例3: CompactorListener

//设置package包名称以及导入依赖的类
package com.groupon.dse.mezzanine.compactor

import com.groupon.dse.mezzanine.partitioner.Partitioner
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.io.Writable
import org.apache.spark.SparkContext
import org.apache.spark.groupon.metrics.UserMetricsSystem
import org.apache.spark.streaming.scheduler.{StreamingListener, StreamingListenerBatchCompleted}


class CompactorListener[K <: Writable, V <: Writable](val sparkContext: SparkContext,
                                                      val fs: FileSystem,
                                                      val partitioner: Partitioner,
                                                      val compactor: Compactor[K, V]) extends StreamingListener {

  override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
    // Get the leaf directories where staging files are written to for the topics we consume
    val stagingLeafDirs = partitioner.stagingLeafDirectories(fs).filter(path => {
      path.getName.startsWith(Partitioner.KeyPrefix)
    })

    stagingLeafDirs.par.foreach(stagingPath => {
      val key = partitioner.keyForStagingDirectory(stagingPath)
      val filesToCompact = compactor.getFilesToCompact(stagingPath, batchCompleted.batchInfo.processingEndTime.get)
      if (filesToCompact.nonEmpty) {
        val outputPath = partitioner.outputDirectory(key)
        UserMetricsSystem.timer(s"mezzanine.write.time.output.$key").time({
          compactor.saveToOutputDir(filesToCompact, outputPath)
        })
        filesToCompact.foreach(fs.delete(_, false))
      }
    })
  }
}

开发者ID:groupon，项目名称:mezzanine，代码行数:35，代码来源:CompactorListener.scala

注：本文中的org.apache.spark.streaming.scheduler.StreamingListenerBatchCompleted类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。