ProgressReporter Contract-spark技术分享

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.

Method Description

currentBatchId



currentBatchId: Long

currentBatchId: Long

Id of the current batch

id



id: UUID

id: UUID

UUID of…FIXME

lastExecution



lastExecution: QueryExecution

lastExecution: QueryExecution

QueryExecution of the streaming query

logicalPlan



logicalPlan: LogicalPlan

logicalPlan: LogicalPlan

The logical query plan of the streaming query

Used when ProgressReporter is requested for the following:

extract statistics from the most recent query execution (to add watermark metric when a streaming watermark is used)
extractSourceToNumInputRows

name



name: String

name: String

newData



newData: Map[BaseStreamingSource, LogicalPlan]

newData: Map[BaseStreamingSource, LogicalPlan]

Streaming sources with the new data as a DataFrame.

Used when:

ProgressReporter extracts statistics from the most recent query execution (to calculate the so-called inputRows)

offsetSeqMetadata



offsetSeqMetadata: OffsetSeqMetadata

offsetSeqMetadata: OffsetSeqMetadata

postEvent



postEvent(event: StreamingQueryListener.Event): Unit

postEvent(event: StreamingQueryListener.Event): Unit

runId



runId: UUID

runId: UUID

sink



sink: BaseStreamingSink

sink: BaseStreamingSink

sources



sources: Seq[BaseStreamingSource]

sources: Seq[BaseStreamingSource]

sparkSession



sparkSession: SparkSession

sparkSession: SparkSession

triggerClock



triggerClock: Clock

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

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

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}

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: 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(newProgress: StreamingQueryProgress): Unit

updateProgress records the input newProgress and posts a QueryProgressEvent event.

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]

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

startTrigger(): Unit

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



DEBUG StreamExecution: Starting Trigger Calculation

DEBUG StreamExecution: Starting Trigger Calculation

startTrigger sets lastTriggerStartTimestamp as currentTriggerStartTimestamp.

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

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]

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[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

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(message: String): Unit

updateStatusMessage updates message in StreamingQueryStatus internal registry.

`extractSourceToNumInputRows` Internal Method



extractSourceToNumInputRows(): Map[BaseStreamingSource, Long]

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(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(hasNewData: Boolean): Seq[StateOperatorProgress]

extractStateOperatorMetrics…FIXME

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

ProgressReporter Contract