An Overview of All of the New Structured Streaming Options Developed In 2021 For Databricks & Apache Spark.

As we enter 2022, we wish to take a second to mirror on the good strides made on the streaming entrance in Databricks and Apache Spark™ ! In 2021, the engineering crew and open supply contributors made quite a few developments with three targets in thoughts:

1. Decrease latency and enhance stateful stream processing
2. Enhance observability of Databricks and Spark Structured Streaming workloads
3. Enhance useful resource allocation and scalability

In the end, the motivation behind these targets was to allow extra groups to run streaming workloads on Databricks and Spark, make it simpler for purchasers to function mission vital manufacturing streaming functions on Databricks and concurrently optimizing for price effectiveness and useful resource utilization.

Aim # 1: Decrease latency & improved stateful processing

There are two new key options that particularly goal reducing latencies with stateful operations, in addition to enhancements to the stateful APIs. The primary is asynchronous checkpointing for giant stateful operations, which improves upon a traditionally synchronous and better latency design.

Asynchronous Checkpointing

On this mannequin, state updates are written to a cloud storage checkpoint location earlier than the following microbatch begins. The benefit is that if a stateful streaming question fails, we will simply restart the question through the use of the data from the final efficiently accomplished batch. Within the asynchronous mannequin, the following microbatch doesn’t have to attend for state updates to be written, bettering the end-to-end latency of the general microbatch execution.

You possibly can study extra about this characteristic in an upcoming deep-dive weblog put up, and check out it in Databricks Runtime 10.3 and above.

Arbitrary stateful operator enhancements

In a a lot earlier put up, we launched Arbitrary Stateful Processing in Structured Streaming with [flat]MapGroupsWithState. These operators present quite a lot of flexibility and allow extra superior stateful operations past aggregations. We’ve launched enhancements to those operators that:

• Permit preliminary state, avoiding the necessity to reprocess all of your streaming information.
• Allow simpler logic testing by exposing a brand new TestGroupState interface, permitting customers to create situations of GroupState and entry inside values for what has been set, simplifying unit checks for the state transition features.

Permit Preliminary State

Let’s begin with the next flatMapGroupswithState operator:

def flatMapGroupsWithState[S: Encoder, U: Encoder](
    outputMode: OutputMode,
    timeoutConf: GroupStateTimeout,
    initialState: KeyValueGroupedDataset[K, S])(
    func: (Okay, Iterator[V], GroupState[S]) => Iterator[U])

This practice state operate maintains a operating rely of fruit which have been encountered.
val fruitCountFunc =(key: String, values: Iterator[String], state: GroupState[RunningCount]) => {
  val rely = state.getOption.map(_.rely).getOrElse(0L) + valList.dimension
  state.replace(new RunningCount(rely))
  Iterator((key, rely.toString))
}

On this instance, we specify the preliminary state to the this operator by setting beginning values for sure fruit:
val fruitCountInitialDS: Dataset[(String, RunningCount)] = Seq(
  ("apple", new RunningCount(1)),
  ("orange", new RunningCount(2)),
  ("mango", new RunningCount(5)),
).toDS()

val fruitCountInitial = initialState.groupByKey(x => x._1).mapValues(_._2)

fruitStream
  .groupByKey(x => x)
  .flatMapGroupsWithState(Replace, GroupStateTimeout.NoTimeout, fruitCountInitial)(fruitCountFunc)

Simpler Logic Testing
It’s also possible to now check state updates utilizing the TestGroupState API.
import org.apache.spark.sql.streaming._
import org.apache.spark.api.java.Non-obligatory

check("flatMapGroupsWithState's state replace operate") {
  var prevState = TestGroupState.create[UserStatus](
    optionalState = Non-obligatory.empty[UserStatus],
    timeoutConf = GroupStateTimeout.EventTimeTimeout,
    batchProcessingTimeMs = 1L,
    eventTimeWatermarkMs = Non-obligatory.of(1L),
    hasTimedOut = false)

  val userId: String = ...
  val actions: Iterator[UserAction] = ...

  assert(!prevState.hasUpdated)

  updateState(userId, actions, prevState)

  assert(prevState.hasUpdated)
  
}

You could find these, and extra examples within the Databricks documentation.
Native help for Session Home windows
Structured Streaming launched the power to do aggregations over event-time based mostly home windows utilizing tumbling or sliding home windows, each of that are home windows of fixed-length.  In Spark 3.2, we launched the idea of session home windows, which permit dynamic window lengths.  This traditionally required customized state operators utilizing flatMapGroupsWithState.
An instance of utilizing dynamic gaps:
# Outline the session window having dynamic hole period based mostly on eventType
session_window expr = session_window(occasions.timestamp, 
    when(occasions.eventType == "type1", "5 seconds") 
    .when(occasions.eventType == "type2", "20 seconds") 
    .in any other case("5 minutes"))

# Group the info by session window and userId, and compute the rely of every group
windowedCountsDF = occasions 
    .withWatermark("timestamp", "10 minutes") 
    .groupBy(occasions.userID, session_window_expr) 
    .rely()

Aim #2: Enhance observability of streaming workloads
Whereas the StreamingQueryListener API permits you to asynchronously monitor queries inside a SparkSession and outline customized callback features for question state, progress, and terminated occasions, understanding again strain and reasoning about the place the bottlenecks are in a microbatch had been nonetheless difficult.  As of Databricks Runtime 8.1, the StreamingQueryProgress object reviews information supply particular again strain metrics for Kafka, Kinesis, Delta Lake and Auto Loader streaming sources.
An instance of the metrics offered for Kafka:
{
  "sources" : [ {
    "description" : "KafkaV2[Subscribe[topic]]",
    "metrics" : {
      "avgOffsetsBehindLatest" : "4.0",
      "maxOffsetsBehindLatest" : "4",
      "minOffsetsBehindLatest" : "4",
      "estimatedTotalBytesBehindLatest" : "80.0"
    },
  } ]
}

Databricks Runtime 8.3 introduces real-time metrics to assist perceive the efficiency of the RocksDB state retailer and debug the efficiency of state operations.  These also can assist establish goal workloads for asynchronous checkpointing.
An instance of the brand new state retailer metrics:
{
  "id" : "6774075e-8869-454b-ad51-513be86cfd43",
  "runId" : "3d08104d-d1d4-4d1a-b21e-0b2e1fb871c5",
  "batchId" : 7,
  "stateOperators" : [ {
    "numRowsTotal" : 20000000,
    "numRowsUpdated" : 20000000,
    "memoryUsedBytes" : 31005397,
    "numRowsDroppedByWatermark" : 0,
    "customMetrics" : {
      "rocksdbBytesCopied" : 141037747,
      "rocksdbCommitCheckpointLatency" : 2,
      "rocksdbCommitCompactLatency" : 22061,
      "rocksdbCommitFileSyncLatencyMs" : 1710,
      "rocksdbCommitFlushLatency" : 19032,
      "rocksdbCommitPauseLatency" : 0,
      "rocksdbCommitWriteBatchLatency" : 56155,
      "rocksdbFilesCopied" : 2,
      "rocksdbFilesReused" : 0,
      "rocksdbGetCount" : 40000000,
      "rocksdbGetLatency" : 21834,
      "rocksdbPutCount" : 1,
      "rocksdbPutLatency" : 56155599000,
      "rocksdbReadBlockCacheHitCount" : 1988,
      "rocksdbReadBlockCacheMissCount" : 40341617,
      "rocksdbSstFileSize" : 141037747,
      "rocksdbTotalBytesReadByCompaction" : 336853375,
      "rocksdbTotalBytesReadByGet" : 680000000,
      "rocksdbTotalBytesReadThroughIterator" : 0,
      "rocksdbTotalBytesWrittenByCompaction" : 141037747,
      "rocksdbTotalBytesWrittenByPut" : 740000012,
      "rocksdbTotalCompactionLatencyMs" : 21949695000,
      "rocksdbWriterStallLatencyMs" : 0,
      "rocksdbZipFileBytesUncompressed" : 7038
    }
  } ],
  "sources" : [ {
  } ],
  "sink" : {
  }
}

Aim # 3: Enhance useful resource allocation and scalability
Streaming Autoscaling with Delta Stay Tables (DLT)
At Knowledge + AI Summit final 12 months, we introduced Delta Stay Tables, which is a framework that permits you to declaratively construct and orchestrate information pipelines, and largely abstracts the necessity to configure clusters and node varieties.   We’re taking this a step additional and introducing an clever autoscaling resolution for streaming pipelines that improves upon the prevailing Databricks Optimized Autoscaling.  These advantages embody:

Higher Cluster Utilization:
The brand new algorithm takes benefit of the brand new again strain metrics to regulate cluster sizes to higher deal with eventualities wherein there are fluctuations in streaming workloads, which in the end results in higher cluster utilization.
Proactive Sleek Employee Shutdown:
Whereas the prevailing autoscaling resolution retires nodes provided that they’re idle, the brand new DLT Autoscaler will proactively shut down chosen nodes when utilization is low, whereas concurrently guaranteeing that there will likely be no failed duties because of the shutdown.


As of writing, this characteristic is at present in Personal Preview.  Please attain out to your account crew for extra data.
Set off.AvailableNow
In Structured Streaming, triggers permit a consumer to outline the timing of a streaming question’s information processing.  These set off varieties could be micro-batch (default), mounted interval micro-batch (Set off.ProcessingTime(“”), one-time micro-batch (Set off.As soon as), and steady (Set off.Steady).
Databricks Runtime 10.1 introduces a brand new sort of set off; Set off.AvailableNow that’s much like Set off.As soon as however offers higher scalability. Like Set off As soon as, all out there information will likely be processed earlier than the question is stopped, however in a number of batches as an alternative of 1. That is supported for Delta Lake and Auto Loader streaming sources.
Instance:
spark.readStream
  .format("delta")
  .possibility("maxFilesPerTrigger", "1")
  .load(inputDir)
  .writeStream
  .set off(Set off.AvailableNow)
  .possibility("checkpointLocation", checkpointDir)
  .begin()

Abstract
As we head into 2022, we are going to proceed to speed up innovation in Structured Streaming, additional bettering efficiency, lowering latency and implementing new and thrilling options. Keep tuned for extra data all year long!