Azure Databricks で RocksDB 状態ストアを構成する

ストリーミング クエリを開始する前に、SparkSession で次の構成を設定することにより、RocksDB ベースの状態管理を有効にすることができます。

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

Delta Live Tables パイプラインで RocksDB を有効にすることができます。 ステートフル処理については、 パイプライン構成の最適化に関するを参照してください。

変更ログのチェックポイント処理を有効にする

Databricks Runtime 13.3 LTS 以降では、構造化ストリーミング ワークロードのチェックポイント期間とエンドツーエンドの待機時間を短縮するために、変更ログのチェックポイント処理を有効にすることができます。 Databricks では、すべての構造化ストリーミング ステートフル クエリに対して変更ログのチェックポイント処理を有効にすることをお勧めします。

従来、RocksDB 状態ストアはチェックポイント処理中にスナップショットを作成し、データ ファイルをアップロードします。 このコストを回避するために、変更ログのチェックポイント処理では、最後のチェックポイント以降に変更されたレコードのみが永続ストレージに書き込まれます。

変更ログのチェックポイント処理は、既定で無効になっています。 SparkSession レベルで変更ログのチェックポイント処理を有効にするには、次の構文を使用します。

spark.conf.set(
  "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled", "true")

既存のストリームで変更ログのチェックポイント処理を有効にし、チェックポイントに格納されている状態情報を維持することができます。

RocksDB 状態ストア メトリクス

各状態操作は、状態ストアを観察するために RocksDB インスタンスで実行された状態管理操作に関連するメトリックを収集します。これは、デバッグ ジョブの低速化につながる可能性があります。 これらのメトリックは、状態操作が実行されているすべてのタスクについて、ジョブの状態操作ごとに集計 (sum) されます。 これらのメトリックは、StreamingQueryProgress の stateOperators フィールド内の customMetrics マップの一部です。 JSON 形式の StreamingQueryProgress の例を次に示します (StreamingQueryProgress.json() を使用して取得)。

{
  "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" : {
  }
}

メトリックの詳細な説明は次のとおりです。