spark JavaStreamingListener 源码

spark JavaStreamingListener 代码

文件路径：/streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListener.scala

/*
 * 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.streaming.api.java

import org.apache.spark.streaming.Time

private[streaming] trait PythonStreamingListener {

  /** Called when the streaming has been started */
  def onStreamingStarted(streamingStarted: JavaStreamingListenerStreamingStarted): Unit = { }

  /** Called when a receiver has been started */
  def onReceiverStarted(receiverStarted: JavaStreamingListenerReceiverStarted): Unit = { }

  /** Called when a receiver has reported an error */
  def onReceiverError(receiverError: JavaStreamingListenerReceiverError): Unit = { }

  /** Called when a receiver has been stopped */
  def onReceiverStopped(receiverStopped: JavaStreamingListenerReceiverStopped): Unit = { }

  /** Called when a batch of jobs has been submitted for processing. */
  def onBatchSubmitted(batchSubmitted: JavaStreamingListenerBatchSubmitted): Unit = { }

  /** Called when processing of a batch of jobs has started.  */
  def onBatchStarted(batchStarted: JavaStreamingListenerBatchStarted): Unit = { }

  /** Called when processing of a batch of jobs has completed. */
  def onBatchCompleted(batchCompleted: JavaStreamingListenerBatchCompleted): Unit = { }

  /** Called when processing of a job of a batch has started. */
  def onOutputOperationStarted(
      outputOperationStarted: JavaStreamingListenerOutputOperationStarted): Unit = { }

  /** Called when processing of a job of a batch has completed. */
  def onOutputOperationCompleted(
      outputOperationCompleted: JavaStreamingListenerOutputOperationCompleted): Unit = { }
}

private[streaming] class PythonStreamingListenerWrapper(listener: PythonStreamingListener)
  extends JavaStreamingListener {

  /** Called when the streaming has been started */
  override def onStreamingStarted(streamingStarted: JavaStreamingListenerStreamingStarted): Unit = {
    listener.onStreamingStarted(streamingStarted)
  }

  /** Called when a receiver has been started */
  override def onReceiverStarted(receiverStarted: JavaStreamingListenerReceiverStarted): Unit = {
    listener.onReceiverStarted(receiverStarted)
  }

  /** Called when a receiver has reported an error */
  override def onReceiverError(receiverError: JavaStreamingListenerReceiverError): Unit = {
    listener.onReceiverError(receiverError)
  }

  /** Called when a receiver has been stopped */
  override def onReceiverStopped(receiverStopped: JavaStreamingListenerReceiverStopped): Unit = {
    listener.onReceiverStopped(receiverStopped)
  }

  /** Called when a batch of jobs has been submitted for processing. */
  override def onBatchSubmitted(batchSubmitted: JavaStreamingListenerBatchSubmitted): Unit = {
    listener.onBatchSubmitted(batchSubmitted)
  }

  /** Called when processing of a batch of jobs has started.  */
  override def onBatchStarted(batchStarted: JavaStreamingListenerBatchStarted): Unit = {
    listener.onBatchStarted(batchStarted)
  }

  /** Called when processing of a batch of jobs has completed. */
  override def onBatchCompleted(batchCompleted: JavaStreamingListenerBatchCompleted): Unit = {
    listener.onBatchCompleted(batchCompleted)
  }

  /** Called when processing of a job of a batch has started. */
  override def onOutputOperationStarted(
    outputOperationStarted: JavaStreamingListenerOutputOperationStarted): Unit = {
      listener.onOutputOperationStarted(outputOperationStarted)
  }

  /** Called when processing of a job of a batch has completed. */
  override def onOutputOperationCompleted(
    outputOperationCompleted: JavaStreamingListenerOutputOperationCompleted): Unit = {
      listener.onOutputOperationCompleted(outputOperationCompleted)
  }
}

/**
 * A listener interface for receiving information about an ongoing streaming  computation.
 */
private[streaming] class JavaStreamingListener {

  /** Called when the streaming has been started */
  def onStreamingStarted(streamingStarted: JavaStreamingListenerStreamingStarted): Unit = { }

  /** Called when a receiver has been started */
  def onReceiverStarted(receiverStarted: JavaStreamingListenerReceiverStarted): Unit = { }

  /** Called when a receiver has reported an error */
  def onReceiverError(receiverError: JavaStreamingListenerReceiverError): Unit = { }

  /** Called when a receiver has been stopped */
  def onReceiverStopped(receiverStopped: JavaStreamingListenerReceiverStopped): Unit = { }

  /** Called when a batch of jobs has been submitted for processing. */
  def onBatchSubmitted(batchSubmitted: JavaStreamingListenerBatchSubmitted): Unit = { }

  /** Called when processing of a batch of jobs has started.  */
  def onBatchStarted(batchStarted: JavaStreamingListenerBatchStarted): Unit = { }

  /** Called when processing of a batch of jobs has completed. */
  def onBatchCompleted(batchCompleted: JavaStreamingListenerBatchCompleted): Unit = { }

  /** Called when processing of a job of a batch has started. */
  def onOutputOperationStarted(
      outputOperationStarted: JavaStreamingListenerOutputOperationStarted): Unit = { }

  /** Called when processing of a job of a batch has completed. */
  def onOutputOperationCompleted(
      outputOperationCompleted: JavaStreamingListenerOutputOperationCompleted): Unit = { }
}

/**
 * Base trait for events related to JavaStreamingListener
 */
private[streaming] sealed trait JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerStreamingStarted(val time: Long)
  extends JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerBatchSubmitted(val batchInfo: JavaBatchInfo)
  extends JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerBatchCompleted(val batchInfo: JavaBatchInfo)
  extends JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerBatchStarted(val batchInfo: JavaBatchInfo)
  extends JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerOutputOperationStarted(
    val outputOperationInfo: JavaOutputOperationInfo) extends JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerOutputOperationCompleted(
    val outputOperationInfo: JavaOutputOperationInfo) extends JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerReceiverStarted(val receiverInfo: JavaReceiverInfo)
  extends JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerReceiverError(val receiverInfo: JavaReceiverInfo)
  extends JavaStreamingListenerEvent

private[streaming] class JavaStreamingListenerReceiverStopped(val receiverInfo: JavaReceiverInfo)
  extends JavaStreamingListenerEvent

/**
 * Class having information on batches.
 *
 * @param batchTime Time of the batch
 * @param streamIdToInputInfo A map of input stream id to its input info
 * @param submissionTime Clock time of when jobs of this batch was submitted to the streaming
 *                       scheduler queue
 * @param processingStartTime Clock time of when the first job of this batch started processing.
 *                            `-1` means the batch has not yet started
 * @param processingEndTime Clock time of when the last job of this batch finished processing. `-1`
 *                          means the batch has not yet completed.
 * @param schedulingDelay Time taken for the first job of this batch to start processing from the
 *                        time this batch was submitted to the streaming scheduler. Essentially, it
 *                        is `processingStartTime` - `submissionTime`. `-1` means the batch has not
 *                        yet started
 * @param processingDelay Time taken for the all jobs of this batch to finish processing from the
 *                        time they started processing. Essentially, it is
 *                        `processingEndTime` - `processingStartTime`. `-1` means the batch has not
 *                        yet completed.
 * @param totalDelay Time taken for all the jobs of this batch to finish processing from the time
 *                   they were submitted.  Essentially, it is `processingDelay` + `schedulingDelay`.
 *                   `-1` means the batch has not yet completed.
 * @param numRecords The number of recorders received by the receivers in this batch
 * @param outputOperationInfos The output operations in this batch
 */
private[streaming] case class JavaBatchInfo(
    batchTime: Time,
    streamIdToInputInfo: java.util.Map[Int, JavaStreamInputInfo],
    submissionTime: Long,
    processingStartTime: Long,
    processingEndTime: Long,
    schedulingDelay: Long,
    processingDelay: Long,
    totalDelay: Long,
    numRecords: Long,
    outputOperationInfos: java.util.Map[Int, JavaOutputOperationInfo])

/**
 * Track the information of input stream at specified batch time.
 *
 * @param inputStreamId the input stream id
 * @param numRecords the number of records in a batch
 * @param metadata metadata for this batch. It should contain at least one standard field named
 *                 "Description" which maps to the content that will be shown in the UI.
 * @param metadataDescription description of this input stream
 */
private[streaming] case class JavaStreamInputInfo(
    inputStreamId: Int,
    numRecords: Long,
    metadata: java.util.Map[String, Any],
    metadataDescription: String)

/**
 * Class having information about a receiver
 */
private[streaming] case class JavaReceiverInfo(
    streamId: Int,
    name: String,
    active: Boolean,
    location: String,
    executorId: String,
    lastErrorMessage: String,
    lastError: String,
    lastErrorTime: Long)

/**
 * Class having information on output operations.
 *
 * @param batchTime Time of the batch
 * @param id Id of this output operation. Different output operations have different ids in a batch.
 * @param name The name of this output operation.
 * @param description The description of this output operation.
 * @param startTime Clock time of when the output operation started processing. `-1` means the
 *                  output operation has not yet started
 * @param endTime Clock time of when the output operation started processing. `-1` means the output
 *                operation has not yet completed
 * @param failureReason Failure reason if this output operation fails. If the output operation is
 *                      successful, this field is `null`.
 */
private[streaming] case class JavaOutputOperationInfo(
    batchTime: Time,
    id: Int,
    name: String,
    description: String,
    startTime: Long,
    endTime: Long,
    failureReason: String)

相关信息

spark 源码目录

相关文章

spark JavaDStream 源码

spark JavaDStreamLike 源码

spark JavaInputDStream 源码

spark JavaMapWithStateDStream 源码

spark JavaPairDStream 源码

spark JavaPairInputDStream 源码

spark JavaPairReceiverInputDStream 源码

spark JavaReceiverInputDStream 源码

spark JavaStreamingContext 源码

spark JavaStreamingListenerWrapper 源码