StreamExecution — Base of Streaming Query Executions-spark技术分享

StreamExecution — Base of Streaming Query Executions

StreamExecution is the base of streaming query executions that can execute the structured query continuously on a stream execution thread.

Note	Continuous query, streaming query, continuous Dataset, streaming Dataset are synonyms, and `StreamExecution` uses analyzed logical plan internally to refer to it.

Property Description

logicalPlan



logicalPlan: LogicalPlan

logicalPlan: LogicalPlan

Analyzed logical plan of the streaming query

Note	`logicalPlan` is part of ProgressReporter Contract and the only purpose of the `logicalPlan` property is to change the access level from `protected` to `public`.

Used when StreamExecution is requested to run stream processing

runActivatedStream



runActivatedStream(sparkSessionForStream: SparkSession): Unit

runActivatedStream(sparkSessionForStream: SparkSession): Unit

Runs the activated streaming query

Used exclusively when StreamExecution is requested to run the streaming query (when transitioning from INITIALIZING to ACTIVE state)

Table 2. StreamExecutions
StreamExecution	Description
ContinuousExecution
MicroBatchExecution

StreamExecution is the execution environment of a single continuous query (aka streaming Dataset) that is executed every trigger and in the end adds the results to a sink.

Note	`StreamExecution` corresponds to a single streaming query with one or more streaming sources and exactly one streaming sink.



scala> spark.version
res0: String = 2.3.0-SNAPSHOT

import org.apache.spark.sql.streaming.Trigger
import scala.concurrent.duration._
val q = spark.
  readStream.
  format("rate").
  load.
  writeStream.
  format("console").
  trigger(Trigger.ProcessingTime(10.minutes)).
  start
scala> :type q
org.apache.spark.sql.streaming.StreamingQuery

// Pull out StreamExecution off StreamingQueryWrapper
import org.apache.spark.sql.execution.streaming.{StreamExecution, StreamingQueryWrapper}
val se = q.asInstanceOf[StreamingQueryWrapper].streamingQuery
scala> :type se
org.apache.spark.sql.execution.streaming.StreamExecution

scala> spark.version

res0: String = 2.3.0-SNAPSHOT

import org.apache.spark.sql.streaming.Trigger

import scala.concurrent.duration._

val q = spark.

readStream.

format("rate").

load.

writeStream.

format("console").

trigger(Trigger.ProcessingTime(10.minutes)).

start

scala> :type q

org.apache.spark.sql.streaming.StreamingQuery

// Pull out StreamExecution off StreamingQueryWrapper

import org.apache.spark.sql.execution.streaming.{StreamExecution, StreamingQueryWrapper}

val se = q.asInstanceOf[StreamingQueryWrapper].streamingQuery

scala> :type se

org.apache.spark.sql.execution.streaming.StreamExecution

Figure 1. Creating Instance of StreamExecution

Note	DataStreamWriter describes how the results of executing batches of a streaming query are written to a streaming sink.

StreamExecution starts a thread of execution that runs the streaming query continuously and concurrently (and polls for new records in the streaming data sources to create a batch every trigger).

Figure 2. StreamExecution’s Starting Streaming Query (on Execution Thread)

StreamExecution can be in three states:

INITIALIZED when the instance was created.
ACTIVE when batches are pulled from the sources.
TERMINATED when executing streaming batches has been terminated due to an error, all batches were successfully processed or StreamExecution has been stopped.

StreamExecution is a ProgressReporter and reports status of the streaming query (i.e. when it starts, progresses and terminates) by posting StreamingQueryListener events.

StreamExecution tracks streaming data sources in uniqueSources internal registry.

Figure 3. StreamExecution’s uniqueSources Registry of Streaming Data Sources

StreamExecution collects durationMs for the execution units of streaming batches.

Figure 4. StreamExecution’s durationMs



scala> :type q
org.apache.spark.sql.streaming.StreamingQuery

scala> println(q.lastProgress)
{
  "id" : "03fc78fc-fe19-408c-a1ae-812d0e28fcee",
  "runId" : "8c247071-afba-40e5-aad2-0e6f45f22488",
  "name" : null,
  "timestamp" : "2017-08-14T20:30:00.004Z",
  "batchId" : 1,
  "numInputRows" : 432,
  "inputRowsPerSecond" : 0.9993568953312452,
  "processedRowsPerSecond" : 1380.1916932907347,
  "durationMs" : {
    "addBatch" : 237,
    "getBatch" : 26,
    "getOffset" : 0,
    "queryPlanning" : 1,
    "triggerExecution" : 313,
    "walCommit" : 45
  },
  "stateOperators" : [ ],
  "sources" : [ {
    "description" : "RateSource[rowsPerSecond=1, rampUpTimeSeconds=0, numPartitions=8]",
    "startOffset" : 0,
    "endOffset" : 432,
    "numInputRows" : 432,
    "inputRowsPerSecond" : 0.9993568953312452,
    "processedRowsPerSecond" : 1380.1916932907347
  } ],
  "sink" : {
    "description" : "ConsoleSink[numRows=20, truncate=true]"
  }
}

scala> :type q

org.apache.spark.sql.streaming.StreamingQuery

scala> println(q.lastProgress)

{

"id" : "03fc78fc-fe19-408c-a1ae-812d0e28fcee",

"runId" : "8c247071-afba-40e5-aad2-0e6f45f22488",

"name" : null,

"timestamp" : "2017-08-14T20:30:00.004Z",

"batchId" : 1,

"numInputRows" : 432,

"inputRowsPerSecond" : 0.9993568953312452,

"processedRowsPerSecond" : 1380.1916932907347,

"durationMs" : {

"addBatch" : 237,

"getBatch" : 26,

"getOffset" : 0,

"queryPlanning" : 1,

"triggerExecution" : 313,

"walCommit" : 45

"stateOperators" : [ ],

"sources" : [ {

"description" : "RateSource[rowsPerSecond=1, rampUpTimeSeconds=0, numPartitions=8]",

"startOffset" : 0,

"endOffset" : 432,

"numInputRows" : 432,

"inputRowsPerSecond" : 0.9993568953312452,

"processedRowsPerSecond" : 1380.1916932907347

} ],

"sink" : {

"description" : "ConsoleSink[numRows=20, truncate=true]"

}

StreamExecution uses OffsetSeqLog and BatchCommitLog metadata logs for write-ahead log (to record offsets to be processed) and that have already been processed and committed to a streaming sink, respectively.

Tip	Monitor `offsets` and `commits` metadata logs to know the progress of a streaming query.

StreamExecution delays polling for new data for 10 milliseconds (when no data was available to process in a batch). Use spark.sql.streaming.pollingDelay Spark property to control the delay.

Name Description

availableOffsets

StreamProgress that tracks the offsets that are available to be processed, but have not yet be committed to the sink.

Note	`availableOffsets` is part of the ProgressReporter Contract.

Note	StreamProgress is an enhanced `immutable.Map` from Scala with streaming sources as keys and their Offsets as values.

Set when (in order):

StreamExecution resumes and populates the start offsets with the latest offsets from the offset log that may have already been processed (and committed to the batch commit log so they are used as the current committed offsets)
StreamExecution constructs the next streaming batch (and gets offsets from the sources)

Note

You can see availableOffsets in the DEBUG message in the logs when StreamExecution resumes and populates the start offsets.



DEBUG Resuming at batch [currentBatchId] with committed offsets [committedOffsets] and available offsets [availableOffsets]

DEBUG Resuming at batch [currentBatchId] with committed offsets [committedOffsets] and available offsets [availableOffsets]

Used when:

StreamExecution starts running streaming batches for the first time (i.e. current batch id is -1 which is right at the initialization time)
StreamExecution checks whether a new data is available in the sources (and is not recorded in committed offsets)
StreamExecution constructs the next streaming batch (and records offsets in the write-ahead offset log)
StreamExecution runs a streaming batch (and fetches data from the sources that has not been processed yet, i.e. not in committed offsets registry)
StreamExecution finishes running streaming batches when data was available in the sources and the offsets have just been committed to a sink (and being added to committed offsets registry)
StreamExecution prints out debug information when a streaming query has terminated due to an exception

Note	`availableOffsets` works in tandem with committedOffsets registry.

awaitProgressLock

Java’s fair reentrant mutual exclusion java.util.concurrent.locks.ReentrantLock (that favors granting access to the longest-waiting thread under contention).

awaitProgressLockCondition

callSite

commitLog

CommitLog with commits metadata checkpoint directory for completed streaming batches (with a single file per batch with a file name being the batch id).

Note	Metadata log or metadata checkpoint are synonyms and are often used interchangeably.

Used exclusively by the extensions for the following:

MicroBatchExecution is requested to runActivatedStream when data was available and the offsets need to be committed (and does populateStartOffsets or constructNextBatch)
ContinuousExecution is requested to getStartOffsets, commit, and awaitEpoch

committedOffsets

StreamProgress of the streaming sources and the committed offsets (i.e. processed already).

Note	`committedOffsets` is a part of ProgressReporter Contract.

currentBatchId

Current batch number

-1 when StreamExecution is created
0 when StreamExecution populates start offsets (and OffsetSeqLog is empty, i.e. no offset files in offsets directory in checkpoint)
Incremented when StreamExecution runs streaming batches and finishes a trigger that had data available from sources (right after committing the batch).

id

Unique identifier of the streaming query

Set as the id of streamMetadata when StreamExecution is created.

Note	`id` can get fetched from checkpoint metadata if available and thus recovered when a query is resumed (i.e. restarted after a failure or a planned stop).

initializationLatch

lastExecution

Last IncrementalExecution

newData



newData: Map[BaseStreamingSource, LogicalPlan]

newData: Map[BaseStreamingSource, LogicalPlan]

Registry of the streaming sources (in the logical query plan) that have new data available in the current batch. The new data is a streaming DataFrame.

Note	`newData` is part of the ProgressReporter Contract.

Set exclusively when StreamExecution is requested to requests unprocessed data from streaming sources (while running a single streaming batch).

Used exclusively when StreamExecution replaces StreamingExecutionRelations in a logical query plan with relations with new data (while running a single streaming batch).

noNewData

Flag whether there are any new offsets available for processing or not.

Turned on (i.e. enabled) when constructing the next streaming batch when no new offsets are available.

offsetLog

OffsetSeqLog with offsets metadata checkpoint directory for write-ahead log to record offsets in when ready for processing.

Note	Metadata log or metadata checkpoint are synonyms and are often used interchangeably.

Used when StreamExecution populates the start offsets and constructs the next streaming batch (first to store the current batch’s offsets in a write-ahead log and retrieve the previous batch’s offsets right afterwards).

Note	`StreamExecution` discards offsets from the offset metadata log when the current batch id is above spark.sql.streaming.minBatchesToRetain Spark property (which defaults to `100`).

offsetSeqMetadata

OffsetSeqMetadata

Note	`offsetSeqMetadata` is a part of ProgressReporter Contract.

Initialized with 0 for batchWatermarkMs and batchTimestampMs when StreamExecution is created.
Updated with 0 for batchWatermarkMs and batchTimestampMs and SparkSession with spark.sql.adaptive.enabled disabled when StreamExecution runs streaming batches.
Used in…FIXME
Copied with batchTimestampMs updated with the current time (in milliseconds) when StreamExecution constructs the next streaming batch.

pollingDelayMs

Time delay before polling new data again when no data was available

Set to spark.sql.streaming.pollingDelay Spark property.

Used when StreamExecution has started running streaming batches (and no data was available to process in a trigger).

prettyIdString

Pretty-identified string for identification in logs (with name if defined).



queryName [id = xyz, runId = abc]

[id = xyz, runId = abc]

queryName [id = xyz, runId = abc]

[id = xyz, runId = abc]

resolvedCheckpointRoot

Qualified path of the checkpoint directory (as defined using checkpointRoot when StreamExecution is created).

Note	checkpointRoot is defined using `checkpointLocation` option or spark.sql.streaming.checkpointLocation Spark property with `queryName` option. `checkpointLocation` and `queryName` options are defined when `StreamingQueryManager` creates a streaming query.

Used when creating the path to the checkpoint directory and when StreamExecution finishes running streaming batches.

Used for logicalPlan (while transforming analyzedPlan and planning StreamingRelation logical operators to corresponding StreamingExecutionRelation physical operators with the streaming data sources created passing in the path to sources directory to store checkpointing metadata).

Note

You can see resolvedCheckpointRoot in the INFO message when StreamExecution is started.



INFO StreamExecution: Starting [id] with [resolvedCheckpointRoot] to store the query checkpoint.

INFO StreamExecution: Starting [id] with [resolvedCheckpointRoot] to store the query checkpoint.

Internally, resolvedCheckpointRoot creates a Hadoop org.apache.hadoop.fs.Path for checkpointRoot and makes it qualified.

Note	`resolvedCheckpointRoot` uses `SparkSession` to access `SessionState` for a Hadoop configuration.

runId

Current run id

sources

All streaming Sources in logical query plan (that are the sources from StreamingExecutionRelation).

startLatch

Java’s java.util.concurrent.CountDownLatch with count 1.

Used when StreamExecution is requested to start to pause the main thread until StreamExecution was requested to run the streaming query.

state

Java’s java.util.concurrent.atomic.AtomicReference for the three different states a streaming query execution can be:

INITIALIZING (default)
ACTIVE (after the first execution of runBatches)
TERMINATED

streamDeathCause

StreamingQueryException

streamMetadata

StreamMetadata from the metadata file from checkpoint directory. If the metadata file is not available it is created (with a new random id).

uniqueSources

Unique streaming data sources in a streaming Dataset (after being collected as StreamingExecutionRelation from the corresponding logical query plan).

Note	StreamingExecutionRelation is a leaf logical operator (i.e. `LogicalPlan`) that represents a streaming data source (and corresponds to a single StreamingRelation in analyzed logical query plan of a streaming Dataset).

Used when StreamExecution:

Constructs the next streaming batch (and gets new offsets for every streaming data source)
Stops all streaming data sources

Tip

Enable INFO or DEBUG logging levels for org.apache.spark.sql.execution.streaming.StreamExecution to see what happens inside.

Add the following line to conf/log4j.properties:



log4j.logger.org.apache.spark.sql.execution.streaming.StreamExecution=DEBUG

log4j.logger.org.apache.spark.sql.execution.streaming.StreamExecution=DEBUG

Refer to Logging.

`stop` Method

Caution

FIXME

`stopSources` Internal Method



stopSources(): Unit

stopSources(): Unit

Caution

FIXME

Running Stream Processing — `runStream` Internal Method



runStream(): Unit

runStream(): Unit

runBatches runs streaming batches of data (that are datasets from every streaming source).



import org.apache.spark.sql.streaming.Trigger
import scala.concurrent.duration._

val out = spark.
  readStream.
  text("server-logs").
  writeStream.
  format("console").
  queryName("debug").
  trigger(Trigger.ProcessingTime(10.seconds))
scala> val debugStream = out.start
INFO StreamExecution: Starting debug [id = 8b57b0bd-fc4a-42eb-81a3-777d7ba5e370, runId = 920b227e-6d02-4a03-a271-c62120258cea]. Use file:///private/var/folders/0w/kb0d3rqn4zb9fcc91pxhgn8w0000gn/T/temporary-274f9ae1-1238-4088-b4a1-5128fc520c1f to store the query checkpoint.
debugStream: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@58a5b69c

// Enable the log level to see the INFO and DEBUG messages
// log4j.logger.org.apache.spark.sql.execution.streaming.StreamExecution=DEBUG

17/06/18 21:21:07 INFO StreamExecution: Starting new streaming query.
17/06/18 21:21:07 DEBUG StreamExecution: getOffset took 5 ms
17/06/18 21:21:07 DEBUG StreamExecution: Stream running from {} to {}
17/06/18 21:21:07 DEBUG StreamExecution: triggerExecution took 9 ms
17/06/18 21:21:07 DEBUG StreamExecution: Execution stats: ExecutionStats(Map(),List(),Map())
17/06/18 21:21:07 INFO StreamExecution: Streaming query made progress: {
  "id" : "8b57b0bd-fc4a-42eb-81a3-777d7ba5e370",
  "runId" : "920b227e-6d02-4a03-a271-c62120258cea",
  "name" : "debug",
  "timestamp" : "2017-06-18T19:21:07.693Z",
  "numInputRows" : 0,
  "processedRowsPerSecond" : 0.0,
  "durationMs" : {
    "getOffset" : 5,
    "triggerExecution" : 9
  },
  "stateOperators" : [ ],
  "sources" : [ {
    "description" : "FileStreamSource[file:/Users/jacek/dev/oss/spark/server-logs]",
    "startOffset" : null,
    "endOffset" : null,
    "numInputRows" : 0,
    "processedRowsPerSecond" : 0.0
  } ],
  "sink" : {
    "description" : "org.apache.spark.sql.execution.streaming.ConsoleSink@2460208a"
  }
}
17/06/18 21:21:10 DEBUG StreamExecution: Starting Trigger Calculation
17/06/18 21:21:10 DEBUG StreamExecution: getOffset took 3 ms
17/06/18 21:21:10 DEBUG StreamExecution: triggerExecution took 3 ms
17/06/18 21:21:10 DEBUG StreamExecution: Execution stats: ExecutionStats(Map(),List(),Map())

import org.apache.spark.sql.streaming.Trigger

import scala.concurrent.duration._

val out = spark.

readStream.

text("server-logs").

writeStream.

format("console").

queryName("debug").

trigger(Trigger.ProcessingTime(10.seconds))

scala> val debugStream = out.start

INFO StreamExecution: Starting debug [id = 8b57b0bd-fc4a-42eb-81a3-777d7ba5e370, runId = 920b227e-6d02-4a03-a271-c62120258cea]. Use file:///private/var/folders/0w/kb0d3rqn4zb9fcc91pxhgn8w0000gn/T/temporary-274f9ae1-1238-4088-b4a1-5128fc520c1f to store the query checkpoint.

debugStream: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@58a5b69c

// Enable the log level to see the INFO and DEBUG messages

// log4j.logger.org.apache.spark.sql.execution.streaming.StreamExecution=DEBUG

17/06/18 21:21:07 INFO StreamExecution: Starting new streaming query.

17/06/18 21:21:07 DEBUG StreamExecution: getOffset took 5 ms

17/06/18 21:21:07 DEBUG StreamExecution: Stream running from {} to {}

17/06/18 21:21:07 DEBUG StreamExecution: triggerExecution took 9 ms

17/06/18 21:21:07 DEBUG StreamExecution: Execution stats: ExecutionStats(Map(),List(),Map())

17/06/18 21:21:07 INFO StreamExecution: Streaming query made progress: {

"id" : "8b57b0bd-fc4a-42eb-81a3-777d7ba5e370",

"runId" : "920b227e-6d02-4a03-a271-c62120258cea",

"name" : "debug",

"timestamp" : "2017-06-18T19:21:07.693Z",

"numInputRows" : 0,

"processedRowsPerSecond" : 0.0,

"durationMs" : {

"getOffset" : 5,

"triggerExecution" : 9

"stateOperators" : [ ],

"sources" : [ {

"description" : "FileStreamSource[file:/Users/jacek/dev/oss/spark/server-logs]",

"startOffset" : null,

"endOffset" : null,

"numInputRows" : 0,

"processedRowsPerSecond" : 0.0

} ],

"sink" : {

"description" : "org.apache.spark.sql.execution.streaming.ConsoleSink@2460208a"

}

17/06/18 21:21:10 DEBUG StreamExecution: Starting Trigger Calculation

17/06/18 21:21:10 DEBUG StreamExecution: getOffset took 3 ms

17/06/18 21:21:10 DEBUG StreamExecution: triggerExecution took 3 ms

17/06/18 21:21:10 DEBUG StreamExecution: Execution stats: ExecutionStats(Map(),List(),Map())

Internally, runBatches assigns the group id (to all the Spark jobs started by this thread) as runId (with the group description to display in web UI as getBatchDescriptionString and interruptOnCancel flag enabled).

Note	`runBatches` uses SparkSession to access `SparkContext` and assign the group id. You can find the details on `SparkContext.setJobGroup` method in the Mastering Apache Spark 2 gitbook.

runBatches sets a local property sql.streaming.queryId as id.

runBatches registers a metric source when spark.sql.streaming.metricsEnabled property is enabled (which is disabled by default).

Caution

FIXME Metrics

runBatches notifies StreamingQueryListeners that a streaming query has been started (by posting a QueryStartedEvent with id, runId and name).

StreamingQueryListener onQueryStarted.png

Figure 5. StreamingQueryListener Notified about Query’s Start (onQueryStarted)

runBatches unblocks the main starting thread (by decrementing the count of startLatch that goes to 0 and lets the starting thread continue).

Caution

FIXME A picture with two parallel lanes for the starting thread and daemon one for the query.

runBatches updates the status message to Initializing sources followed by initialization of the logical plan (of the streaming Dataset).

runBatches disables adaptive query execution (using spark.sql.adaptive.enabled property which is disabled by default) as it could change the number of shuffle partitions.

runBatches initializes offsetSeqMetadata internal variable.

runBatches sets state to ACTIVE (only when the current state is INITIALIZING that prevents from repeating the initialization)

Note	`runBatches` does the work only when first started (i.e. when state is `INITIALIZING`).

runBatches decrements the count of initializationLatch.

Caution

FIXME initializationLatch so what?

runBatches runs the activated streaming query.

Once TriggerExecutor has finished executing batches, runBatches updates the status message to Stopped.

Note	TriggerExecutor finishes executing batches when batch runner returns whether the streaming query is stopped or not (which is when the internal state is not `TERMINATED`).

Caution

FIXME Describe catch block for exception handling

Caution

FIXME Describe finally block for query termination

Note	`runStream` is used exclusively when the stream execution thread is requested to start.

TriggerExecutor’s Batch Runner

Batch Runner (aka batchRunner) is an executable block executed by TriggerExecutor in runBatches.

batchRunner starts trigger calculation.

As long as the query is not stopped (i.e. state is not TERMINATED), batchRunner executes the streaming batch for the trigger.

In triggerExecution time-tracking section, runBatches branches off per currentBatchId.

Table 4. Current Batch Execution per currentBatchId

currentBatchId < 0

currentBatchId >= 0

populateStartOffsets
Setting Job Description as getBatchDescriptionString



DEBUG Stream running from [committedOffsets] to [availableOffsets]

DEBUG Stream running from [committedOffsets] to [availableOffsets]

1. Constructing the next streaming batch

If there is data available in the sources, batchRunner marks currentStatus with isDataAvailable enabled.

Note

You can check out the status of a streaming query using status method.



scala> spark.streams.active(0).status
res1: org.apache.spark.sql.streaming.StreamingQueryStatus =
{
  "message" : "Waiting for next trigger",
  "isDataAvailable" : false,
  "isTriggerActive" : false
}

scala> spark.streams.active(0).status

res1: org.apache.spark.sql.streaming.StreamingQueryStatus =

{

"message" : "Waiting for next trigger",

"isDataAvailable" : false,

"isTriggerActive" : false

}

batchRunner then updates the status message to Processing new data and runs the current streaming batch.

Figure 6. StreamExecution’s Running Batches (on Execution Thread)

After triggerExecution section has finished, batchRunner finishes the streaming batch for the trigger (and collects query execution statistics).

When there was data available in the sources, batchRunner updates committed offsets (by adding the current batch id to BatchCommitLog and adding availableOffsets to committedOffsets).

You should see the following DEBUG message in the logs:



DEBUG batch $currentBatchId committed

DEBUG batch $currentBatchId committed

batchRunner increments the current batch id and sets the job description for all the following Spark jobs to include the new batch id.

When no data was available in the sources to process, batchRunner does the following:

Marks currentStatus with isDataAvailable disabled
Updates the status message to Waiting for data to arrive
Sleeps the current thread for pollingDelayMs milliseconds.

batchRunner updates the status message to Waiting for next trigger and returns whether the query is currently active or not (so TriggerExecutor can decide whether to finish executing the batches or not)

`getBatchDescriptionString` Internal Method



getBatchDescriptionString: String

getBatchDescriptionString: String

Caution

FIXME

`toDebugString` Internal Method



toDebugString(includeLogicalPlan: Boolean): String

toDebugString(includeLogicalPlan: Boolean): String

toDebugString…FIXME

Note	`toDebugString` is used exclusively when `StreamExecution` is requested to run stream processing (and a streaming query terminated with an exception).

Starting Streaming Query (on Stream Execution Thread) — `start` Method



start(): Unit

start(): Unit

When called, start prints out the following INFO message to the logs:



Starting [id]. Use [resolvedCheckpointRoot] to store the query checkpoint.

Starting [id]. Use [resolvedCheckpointRoot] to store the query checkpoint.

start then starts the queryExecutionThread as a daemon thread.

Note	`start` uses Java’s java.lang.Thread.start to run the streaming query on a separate execution thread.

Note	When started, a streaming query runs in its own execution thread on JVM.

In the end, start pauses the main thread (using the startLatch until StreamExecution was requested to run the streaming query).

Note	`start` is used exclusively when `StreamingQueryManager` is requested to start a streaming query.

Creating StreamExecution Instance

StreamExecution takes the following when created:

SparkSession
Query name
Path of the checkpoint directory (aka metadata directory)
Analyzed logical query plan (i.e. LogicalPlan)
Streaming sink
Trigger
Clock
Output mode (that is only used when creating IncrementalExecution for a streaming batch in query planning)
deleteCheckpointOnStop flag to control whether to delete the checkpoint directory on stop

StreamExecution initializes the internal registries and counters.

Note	`StreamExecution` is a Scala abstract class and cannot be created directly. It is created indirectly when the concrete StreamExecutions are.

Creating Path to Checkpoint Directory — `checkpointFile` Internal Method



checkpointFile(name: String): String

checkpointFile(name: String): String

checkpointFile gives the path of a directory with name in checkpoint directory.

Note	`checkpointFile` uses Hadoop’s `org.apache.hadoop.fs.Path`.

Note	`checkpointFile` is used for streamMetadata, OffsetSeqLog, BatchCommitLog, and lastExecution (for runBatch).

Posting StreamingQueryListener Event — `postEvent` Method



postEvent(event: StreamingQueryListener.Event): Unit

postEvent(event: StreamingQueryListener.Event): Unit

Note	`postEvent` is a part of ProgressReporter Contract.

postEvent simply requests the StreamingQueryManager to post the input event (to the StreamingQueryListenerBus in the current SparkSession).

Note	`postEvent` uses `SparkSession` to access the current `StreamingQueryManager`.

Note	`postEvent` is used when: `ProgressReporter` reports update progress (while finishing a trigger) `StreamExecution` runs streaming batches (and announces starting a streaming query by posting a QueryStartedEvent and query termination by posting a QueryTerminatedEvent)

Waiting Until No Data Available in Sources or Query Has Been Terminated — `processAllAvailable` Method



processAllAvailable(): Unit

processAllAvailable(): Unit

Note	`processAllAvailable` is a part of StreamingQuery Contract.

processAllAvailable reports streamDeathCause exception if defined (and returns).

Note	streamDeathCause is defined exclusively when `StreamExecution` runs streaming batches (and terminated with an exception).

processAllAvailable returns when isActive flag is turned off (which is when StreamExecution is in TERMINATED state).

processAllAvailable acquires a lock on awaitProgressLock and turns noNewData flag off.

processAllAvailable keeps waiting 10 seconds for awaitProgressLockCondition until noNewData flag is turned on or StreamExecution is no longer active.

Note	noNewData flag is turned on exclusively when `StreamExecution` constructs the next streaming batch (and finds that no data is available).

In the end, processAllAvailable releases awaitProgressLock lock.

Stream Execution Thread — `queryExecutionThread` Property



queryExecutionThread: QueryExecutionThread

queryExecutionThread: QueryExecutionThread

queryExecutionThread is a Java thread of execution (java.util.Thread) that runs the structured query when started.

queryExecutionThread uses the name stream execution thread for [id] (that uses prettyIdString for the id, i.e. queryName [id = [id], runId = [runId]]).

queryExecutionThread is a QueryExecutionThread (that is a Spark UninterruptibleThread with runUninterruptibly method for running a block of code without being interrupted by Thread.interrupt()).

queryExecutionThread is started (as a daemon thread) when StreamExecution is requested to start.

When started, queryExecutionThread sets the thread-local properties as the call site and runs the streaming query.

Tip

Use Java’s jconsole or jstack to monitor the streaming threads.



$ jstack <driver-pid> \| grep -e "stream execution thread"
"stream execution thread for kafka-topic1 [id =...

$ jstack <driver-pid> \| grep -e "stream execution thread"

"stream execution thread for kafka-topic1 [id =...

StreamExecution — Base of Streaming Query Executions