Configurer le magasin d'état RocksDB sur Databricks
RocksDB est le fournisseur de magasin d'état default dans Databricks Runtime 17.3 et les versions ultérieures. Pour les versions de Databricks Runtime antérieures à la 17.3, vous pouvez activer la gestion d'état basée sur RocksDB en définissant la configuration suivante dans la SparkSession avant de démarrer la query de streaming.
spark.conf.set(
"spark.sql.streaming.stateStore.providerClass",
"com.databricks.sql.streaming.state.RocksDBStateStoreProvider")
Vous pouvez activer RocksDB sur les LakeFlow Pipelines. Voir Optimiser la configuration du pipeline pour le traitement avec état.
Activer la journalisation des points de contrôle
Dans Databricks Runtime 13.3 LTS et versions ultérieures, vous pouvez activer le checkpointing de changelog pour réduire la durée du checkpoint et la latence de bout en bout pour les charges de travail Structured Streaming. Databricks recommande d'activer le point de contrôle de changelog pour toutes les query Structured Streaming avec état. Le checkpointing du journal des modifications est activé default dans Databricks Runtime 17.3 et versions ultérieures.
Traditionnellement, RocksDB State Store capture des instantanés et télécharge les fichiers de données lors du point de contrôle. Pour éviter ce coût, le checkpointing des journaux de modifications n'écrit que les enregistrements qui ont changé depuis le dernier point de contrôle dans le stockage durable.
Le point de contrôle du journal des modifications est désactivé by default sur les versions de Databricks Runtime antérieures à 17.3. Vous pouvez activer le checkpointing du journal des modifications au niveau de la SparkSession en utilisant la syntaxe suivante :
spark.conf.set(
"spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled", "true")
Vous pouvez activer les points de contrôle du journal des modifications sur un Stream existant et maintenir les informations d'état stockées dans le point de contrôle.
Les query ayant activé le point de contrôle du journal des modifications ne peuvent être exécutés que sur Databricks Runtime 13.3 LTS ou version ultérieure. Vous pouvez désactiver le point de contrôle du journal des modifications pour revenir au comportement de point de contrôle hérité, mais vous devez continuer à exécuter ces query sur Databricks Runtime 13.3 LTS ou version ultérieure. Vous devez redémarrer le Job pour que ces modifications prennent effet.
Métriques du stockage d'état RocksDB
Chaque opérateur d'état collecte des métriques liées aux opérations de gestion d'état effectuées sur son instance RocksDB pour observer le magasin d'état et potentiellement aider au debugging de la lenteur des Jobs.
Dans Databricks Runtime 16.4 LTS et versions supérieures, les métriques d'une instance de magasin d'état spécifique sont étiquetées avec leur ID de partition et le nom du magasin, garantissant qu'elles restent séparées. Toutes les autres métriques sont rapportées comme la somme agrégée pour chaque opérateur d'état sur toutes les tâches lorsque l'opérateur d'état est en cours d'exécution.
Ces métriques font partie de la carte customMetrics à l'intérieur des champs stateOperators de StreamingQueryProgress. Voici un exemple de StreamingQueryProgress au format JSON (obtenu à l'aide de 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": {}
}
Les descriptions détaillées des métriques sont les suivantes :
Nom de la métrique | Description |
|---|---|
rocksdbCommitWriteBatchLatency | Temps (en millisecondes) nécessaire pour appliquer les écritures étagées dans la structure en mémoire (WriteBatch) à RocksDB natif. |
rocksdbCommitFlushLatency | Temps (en ms) nécessaire pour vider les modifications en mémoire de RocksDB sur le disque local. |
rocksdbCommitCompactLatency | Temps (en millisecondes) nécessaire pour le compactage (facultatif) lors du commit du point de contrôle. |
rocksdbCommitPauseLatency | Temps (en millisecondes) nécessaire à l'arrêt des threads worker en arrière-plan (pour la compaction, etc.) dans le cadre du commit du point de contrôle. |
rocksdbCommitCheckpointLatency | Durée (en millisecondes) nécessaire pour prendre un instantané de RocksDB natif et l’écrire dans un répertoire local. |
rocksdbCommitFileSyncLatencyMs | Temps (en millisecondes) nécessaire pour synchroniser les fichiers liés à l'instantané RocksDB natif vers un stockage externe (emplacement de point de contrôle). |
rocksdbGetLatency | Temps moyen (en nanosecondes) pris par l'appel natif sous-jacent |
rocksdbPutCount | Temps moyen (en nanosecondes) pris par l'appel natif sous-jacent |
rocksdbGetCount | Nombre d'appels natifs |
rocksdbPutCount | Nombre d'appels natifs |
rocksdbTotalBytesReadByGet | Nombre d'octets non compressés lus via les appels natifs |
rocksdbTotalBytesWrittenByPut | Nombre d'octets décompressés écrits via les appels natifs |
rocksdbReadBlockCacheHitCount | Nombre de fois que le cache de blocs natif RocksDB est utilisé pour éviter de lire des données à partir du disque local. |
rocksdbReadBlockCacheMissCount | Nombre de fois où le cache de blocs RocksDB natif a manqué et a nécessité la lecture de données depuis le disque local. |
rocksdbTotalBytesReadByCompaction | Nombre d'octets lus sur le disque local par le processus de compaction natif de RocksDB. |
rocksdbTotalBytesWrittenByCompaction | Nombre d'octets écrits sur le disque local par le processus de compactage natif de RocksDB. |
rocksdbTotalCompactionLatencyMs | Temps (en ms) nécessaire pour les compactages RocksDB (à la fois l'arrière-plan et le compactage facultatif initié pendant le commit). |
rocksdbWriterStallLatencyMs | Temps (en millisecondes) pendant lequel l'enregistreur a été bloqué en raison d'un compactage en arrière-plan ou d'un vidage des memtables sur le disque. |
rocksdbTotalBytesReadThroughIterator | Certaines des opérations avec état (telles que le traitement des délais dans |
Limiter l'utilisation de la mémoire RocksDB
Dans Databricks Runtime 17,3 et versions supérieures, Databricks plafonne automatiquement l'utilisation de la mémoire RocksDB par nœud. Sur les versions antérieures de Databricks Runtime, configurez ceci manuellement pour éviter les erreurs de mémoire insuffisante.
RocksDB alloue de la mémoire pour les memtables, le cache de blocs, ainsi que les blocs de filtre et d'index. Sans limite, l'utilisation de la mémoire par plusieurs instances RocksDB sur un nœud peut augmenter indéfiniment et provoquer des erreurs de mémoire. Le gestionnaire de tampon d'écriture RocksDB applique une limite de mémoire par nœud à toutes les instances RocksDB exécutées sur le nœud.
Pour plafonner l'utilisation de la mémoire de RocksDB, définissez les configurations suivantes dans la session Spark avant de démarrer la query en streaming :
spark.conf.set(
"spark.sql.streaming.stateStore.rocksdb.boundedMemoryUsage", "true")
spark.conf.set(
"spark.sql.streaming.stateStore.rocksdb.maxMemoryUsageMB", "<value>")
Les configurations suivantes contrôlent l'utilisation de la mémoire RocksDB :
Configuration | Description |
|---|---|
| Active une limite de mémoire partagée pour toutes les instances RocksDB sur un nœud. Définissez-le sur |
| Mémoire maximale en Mo partagée entre toutes les instances RocksDB sur un nœud. Définissez cette valeur statique, ou à une valeur calculée comme une fraction de la mémoire physique du nœud. |
| Taille maximale de la mémoire tampon d'écriture en Mo pour une instance RocksDB individuelle. La valeur par default est la valeur interne de RocksDB. |
| Nombre maximal de tampons d’écriture pour une instance RocksDB individuelle. La valeur par default est la valeur interne de RocksDB. |