Azure Databricks에서 RocksDB 상태 저장소 구성

스트리밍 쿼리를 시작하기 전에 SparkSession에서 다음 구성을 설정하여 RocksDB 기반 상태 관리를 사용하도록 설정할 수 있습니다.

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

Lakeflow Spark 선언적 파이프라인에서 RocksDB를 사용하도록 설정할 수 있습니다. 상태 저장 처리의 파이프라인 구성 최적화 ()를 참조하세요.

변경 로그 검사점 사용

Databricks Runtime 13.3 LTS 이상에서 구조적 스트리밍 워크로드에 대한 검사점 기간 및 엔드 투 엔드 대기 시간을 낮추도록 변경 로그 검사점을 사용하도록 설정할 수 있습니다. Databricks는 모든 구조적 스트리밍 상태 저장 쿼리에 대해 changelog 검사포인트를 사용하도록 설정할 것을 권합니다.

일반적으로 RocksDB 상태 저장소는 검사점이 있는 동안 데이터 파일을 스냅샷하고 업로드합니다. 이 비용을 방지하기 위해 변경 로그 검사점은 마지막 검사점 이후 변경된 레코드만 지속성 스토리지에 기록합니다."

변경 로그 검사점은 기본적으로 사용하지 않도록 설정됩니다. 다음 구문을 사용하여 SparkSession 수준에서 변경 로그 검사점을 사용하도록 설정할 수 있습니다.

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

기존 스트림에서 변경 로그 검사점을 사용하도록 설정하고 검사점에 저장된 상태 정보를 유지할 수 있습니다.

RocksDB 상태 저장소 메트릭

각 상태 연산자는 해당 RocksDB 인스턴스에서 수행된 상태 관리 작업과 관련된 메트릭을 수집하여 상태 저장소를 관찰하고 작업 속도 저하 문제의 디버그를 지원할 수 있습니다.

Databricks Runtime 16.4 LTS 이상에서 특정 상태 저장소 인스턴스에 대한 메트릭은 파티션 ID 및 저장소 이름으로 레이블이 지정되어 별도로 유지됩니다. 다른 모든 메트릭은 상태 연산자가 실행 중인 모든 작업에서 각 상태 연산자의 집계 합계로 보고됩니다.

이러한 메트릭은 customMetrics의 stateOperators 필드 내 StreamingQueryProgress 맵의 일부입니다. 다음은 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": {
        "SnapshotLastUploaded.partition_0_default": 7,
        "SnapshotLastUploaded.partition_1_default": 7,
        "SnapshotLastUploaded.partition_2_default": 6,
        "SnapshotLastUploaded.partition_3_default": 6,
        "SnapshotLastUploaded.partition_4_default": -1,
        "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": {}
}

메트릭에 대한 자세한 설명은 다음과 같습니다.