ProgressReporter Contract

ProgressReporter is the contract of stream execution progress reporters that continually report statistics about the amount of data processed and the latency of a streaming query.

Note
StreamExecution is the one and only known direct extension of the ProgressReporter Contract in Spark Structured Streaming.
Table 1. ProgressReporter Contract
Method Description

currentBatchId

currentBatchId: Long

Id of the current batch

id

id: UUID

UUID of…​FIXME

lastExecution

lastExecution: QueryExecution

QueryExecution of the streaming query

logicalPlan

logicalPlan: LogicalPlan

The logical query plan of the streaming query

Used when ProgressReporter is requested for the following:

name

name: String

newData

newData: Map[BaseStreamingSource, LogicalPlan]

Streaming sources with the new data as a DataFrame.

Used when:

offsetSeqMetadata

offsetSeqMetadata: OffsetSeqMetadata

postEvent

postEvent(event: StreamingQueryListener.Event): Unit

runId

runId: UUID

sink

sink: BaseStreamingSink

sources

sources: Seq[BaseStreamingSource]

sparkSession

sparkSession: SparkSession

triggerClock

triggerClock: Clock

The clock to use to track the time

import org.apache.spark.sql.streaming.Trigger
import scala.concurrent.duration._
val sampleQuery = spark
  .readStream
  .format("rate")
  .load
  .writeStream
  .format("console")
  .option("truncate", false)
  .trigger(Trigger.ProcessingTime(10.seconds))
  .start

// Using public API
import org.apache.spark.sql.streaming.SourceProgress
scala> sampleQuery.
     |   lastProgress.
     |   sources.
     |   map { case sp: SourceProgress =>
     |     s"source = ${sp.description} => endOffset = ${sp.endOffset}" }.
     |   foreach(println)
source = RateSource[rowsPerSecond=1, rampUpTimeSeconds=0, numPartitions=8] => endOffset = 663

scala> println(sampleQuery.lastProgress.sources(0))
res40: org.apache.spark.sql.streaming.SourceProgress =
{
  "description" : "RateSource[rowsPerSecond=1, rampUpTimeSeconds=0, numPartitions=8]",
  "startOffset" : 333,
  "endOffset" : 343,
  "numInputRows" : 10,
  "inputRowsPerSecond" : 0.9998000399920015,
  "processedRowsPerSecond" : 200.0
}

// With a hack
import org.apache.spark.sql.execution.streaming.StreamingQueryWrapper
val offsets = sampleQuery.
  asInstanceOf[StreamingQueryWrapper].
  streamingQuery.
  availableOffsets.
  map { case (source, offset) =>
    s"source = $source => offset = $offset" }
scala> offsets.foreach(println)
source = RateSource[rowsPerSecond=1, rampUpTimeSeconds=0, numPartitions=8] => offset = 293
Table 2. ProgressReporter’s Internal Properties (e.g. Registries, Counters and Flags)
Name Description

currentDurationsMs

scala.collection.mutable.HashMap of action names (aka triggerDetailKey) and their cumulative times (in milliseconds).

The action names can be as follows:

  • addBatch

  • getBatch (when StreamExecution runs a streaming batch)

  • getOffset

  • queryPlanning

  • triggerExecution

  • walCommit when writing offsets to log

Starts empty when ProgressReporter sets the state for a new batch with new entries added or updated when reporting execution time (of an action).

Tip

You can see the current value of currentDurationsMs in progress reports under durationMs.

scala> query.lastProgress.durationMs
res3: java.util.Map[String,Long] = {triggerExecution=60, queryPlanning=1, getBatch=5, getOffset=0, addBatch=30, walCommit=23}

currentStatus

StreamingQueryStatus to track the status of a streaming query.

Available using status method.

currentTriggerEndTimestamp

Timestamp of when the current batch/trigger has ended

currentTriggerStartTimestamp

Timestamp of when the current batch/trigger has started

noDataProgressEventInterval

FIXME

lastNoDataProgressEventTime

FIXME

lastTriggerStartTimestamp

Timestamp of when the last batch/trigger started

progressBuffer

scala.collection.mutable.Queue of StreamingQueryProgress

Elements are added and removed when ProgressReporter updates progress.

Used when ProgressReporter does lastProgress or recentProgress.

SourceProgress

Caution
FIXME

SinkProgress

Caution
FIXME

status Method

status: StreamingQueryStatus

status gives the current StreamingQueryStatus.

Note
status is used when StreamingQueryWrapper is requested for the current status of a streaming query (that is part of StreamingQuery Contract).

Reporting Streaming Query Progress — updateProgress Internal Method

updateProgress(newProgress: StreamingQueryProgress): Unit

updateProgress records the input newProgress and posts a QueryProgressEvent event.

ProgressReporter updateProgress.png
Figure 1. ProgressReporter’s Reporting Query Progress

updateProgress adds the input newProgress to progressBuffer.

updateProgress removes elements from progressBuffer if their number is or exceeds the value of spark.sql.streaming.numRecentProgressUpdates property.

updateProgress posts a QueryProgressEvent (with the input newProgress).

updateProgress prints out the following INFO message to the logs:

INFO StreamExecution: Streaming query made progress: [newProgress]
Note
updateProgress synchronizes concurrent access to progressBuffer.
Note
updateProgress is used exclusively when ProgressReporter finishes a trigger.

Setting State For New Trigger — startTrigger Method

startTrigger(): Unit

When called, startTrigger prints out the following DEBUG message to the logs:

DEBUG StreamExecution: Starting Trigger Calculation

startTrigger sets currentTriggerStartTimestamp using triggerClock.

startTrigger enables isTriggerActive flag of StreamingQueryStatus.

startTrigger clears currentDurationsMs.

Note
startTrigger is used exclusively when StreamExecution starts running batches (as part of TriggerExecutor executing a batch runner).

Finishing Trigger (Updating Progress and Marking Current Status As Trigger Inactive) — finishTrigger Method

finishTrigger(hasNewData: Boolean): Unit

Internally, finishTrigger sets currentTriggerEndTimestamp to the current time (using triggerClock).

finishTrigger extractExecutionStats.

finishTrigger calculates the processing time (in seconds) as the difference between the end and start timestamps.

finishTrigger calculates the input time (in seconds) as the difference between the start time of the current and last triggers.

ProgressReporter finishTrigger timestamps.png
Figure 2. ProgressReporter’s finishTrigger and Timestamps

finishTrigger prints out the following DEBUG message to the logs:

Execution stats: [executionStats]

finishTrigger creates a SourceProgress (aka source statistics) for every source used.

finishTrigger creates a SinkProgress (aka sink statistics) for the sink.

finishTrigger creates a StreamingQueryProgress.

If there was any data (using the input hasNewData flag), finishTrigger resets lastNoDataProgressEventTime (i.e. becomes the minimum possible time) and updates query progress.

Otherwise, when no data was available (using the input hasNewData flag), finishTrigger updates query progress only when lastNoDataProgressEventTime passed.

In the end, finishTrigger disables isTriggerActive flag of StreamingQueryStatus (i.e. sets it to false).

Note
finishTrigger is used exclusively when MicroBatchExecution is requested to run the activated streaming query.

Tracking and Recording Execution Time — reportTimeTaken Method

reportTimeTaken[T](triggerDetailKey: String)(body: => T): T

reportTimeTaken measures the time to execute body and records it in currentDurationsMs.

In the end, reportTimeTaken prints out the following DEBUG message to the logs and returns the result of executing body.

DEBUG StreamExecution: [triggerDetailKey] took [time] ms
Note

reportTimeTaken is used when StreamExecution wants to record the time taken for (as triggerDetailKey in the DEBUG message above):

  • addBatch

  • getBatch

  • getOffset

  • queryPlanning

  • triggerExecution

  • walCommit when writing offsets to log

updateStatusMessage Method

updateStatusMessage(message: String): Unit

updateStatusMessage updates message in StreamingQueryStatus internal registry.

extractSourceToNumInputRows Internal Method

extractSourceToNumInputRows(): Map[BaseStreamingSource, Long]

extractSourceToNumInputRows…​FIXME

Note
extractSourceToNumInputRows is used exclusively when ProgressReporter is requested to extractExecutionStats.

Creating Execution Statistics — extractExecutionStats Internal Method

extractExecutionStats(hasNewData: Boolean): ExecutionStats

extractExecutionStats…​FIXME

Note
extractExecutionStats is used exclusively when ProgressReporter is requested to finishTrigger.

extractStateOperatorMetrics Internal Method

extractStateOperatorMetrics(hasNewData: Boolean): Seq[StateOperatorProgress]

extractStateOperatorMetrics…​FIXME

Note
extractStateOperatorMetrics is used exclusively when ProgressReporter is requested to extractExecutionStats.

results matching ""

    No results matching ""