有状态查询的异步状态检查点

异步状态检查点为流式处理查询维护恰好一次的保证，但可以为在状态更新上出现瓶颈的一些结构化流式处理有状态工作负载减少总体延迟。 这通过在完成上一个微批处理的计算后立即开始处理下一个微批处理来实现，无需等待状态检查点完成。 下表比较了同步检查点和异步检查点的权衡：

以下是可能从异步状态检查点中受益的流式处理作业的特征：

• 作业有一个或多个有状态操作（例如，聚合、flatMapGroupsWithState、mapGroupsWithState、流之间的联接）
• 状态检查点延迟是导致整体批处理执行延迟的主要因素之一。 此信息可以在 StreamingQueryProgress 事件中找到。 这些事件也可在 Spark 驱动程序上的 log4j 日志中找到。 下面是流式处理查询进度的示例，以及如何查找状态检查点对总体批处理执行延迟的影响。

  • {
       "id" : "2e3495a2-de2c-4a6a-9a8e-f6d4c4796f19",
       "runId" : "e36e9d7e-d2b1-4a43-b0b3-e875e767e1fe",
       "...",
       "batchId" : 0,
       "durationMs" : {
         "...",
         "triggerExecution" : 547730,
         "..."
       },
       "stateOperators" : [ {
         "...",
         "commitTimeMs" : 3186626,
         "numShufflePartitions" : 64,
         "..."
       }]
    }
    

  • 上述查询进度事件的状态检查点延迟分析

    • 批处理持续时间 (durationMs.triggerDuration) 约为 547 秒。
    • 状态存储提交延迟时间 (stateOperations[0].commitTimeMs) 大约为 3186 秒。 提交延迟时间是所有包含状态存储的任务的总和。 在此例中共有 64 个这样的任务 (stateOperators[0].numShufflePartitions)。
    • 每个包含状态运算符的任务平均花费 50 秒（3186/64）用于检查点。 这导致了批处理持续时间的额外延迟。 假设所有 64 个任务并发运行，则检查点步骤占据了大约 9%（50 秒/547 秒）的批处理持续时间。 当并发运行的最大任务数小于 64 时，这一百分比会更高。

启用异步状态检查点

必须为异步状态检查点使用基于 RocksDB 的状态存储。 设置以下配置：

spark.conf.set(
  "spark.databricks.streaming.statefulOperator.asyncCheckpoint.enabled",
  "true"
)

spark.conf.set(
  "spark.sql.streaming.stateStore.providerClass",
  "com.databricks.sql.streaming.state.RocksDBStateStoreProvider"
)

异步检查点的限制和要求

• 在任何一个或多个存储内，异步检查点中的任何故障都会导致查询失败。 在同步检查点模式下，检查点作为任务的一部分执行，Spark 在查询失败之前会多次重试任务。 异步状态检查点不提供此机制。 但是，使用 Databricks 作业重试，可以自动重试此类失败。
• 当状态存储位置在微批处理执行之间未更改时，异步检查点最有效。 群集大小重设与异步状态检查可能无法正常结合使用，因为在出现群集大小重设事件时，状态存储实例可能会在添加或删除节点时重新分布。
• 仅在 RocksDB 状态存储提供程序实现中支持异步状态检查点。 默认的内存中状态存储实现不支持异步状态检查点。