Réalisation d’un point de contrôle d’état asynchrone pour des requêtes avec état

La réalisation d’un point de contrôle d’état asynchrone maintient des garanties exactement une fois pour des requêtes de diffuser en continu, mais elle peut réduire la latence globale pour certaines charges de travail avec état Structured Streaming rencontrant un goulot d’étranglement sur des mises à jour d’état. Pour ce faire, commencez à traiter le micro-lot suivant dès que le calcul du micro-lot précédent est terminé sans attendre la fin de la réalisation du point de contrôle d’état. Le tableau suivant compare les compromis pour la réalisation de points de contrôle synchrones et asynchrones :

Voici les caractéristiques du travail de diffusion en continu qui peuvent bénéficier de la réalisation d’un point de contrôle d’état asynchrone :

• Le travail comprend une ou plusieurs opérations avec état (par exemple, agrégation, flatMapGroupsWithState, mapGroupsWithState, jointures flux-flux)
• La latence du point de contrôle d’état est l’un des facteurs majeurs de la latence d’exécution globale du lot. Ces informations se trouvent dans les événements StreamingQueryProgress. Ces événements se trouvent également dans des journaux log4j sur le pilote Spark. Voici un exemple de progression de requête de diffusion en continu et comment trouver l’impact du point de contrôle d’état sur la latence d’exécution globale du lot.

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

  • Analyse de la latence du point de contrôle d’état de l’événement de progression de requête ci-dessus

    • La durée du lot (durationMs.triggerDuration) est d’environ 547 secondes.
    • La latence de validation du magasin d’état (stateOperations[0].commitTimeMs) est d’environ 3 186 secondes. La latence de validation est agrégée entre les tâches contenant un magasin d’état. Dans ce cas, il existe 64 tâches de ce type (stateOperators[0].numShufflePartitions).
    • Chaque tâche contenant un opérateur d’état a pris en moyenne 50 secondes (3 186/64) pour le point de contrôle. Il s’agit d’une latence supplémentaire qui a contribué à la durée du lot. En supposant que les 64 tâches s’exécutent simultanément, l’étape de point de contrôle a contribué à environ 9 % (50 secondes / 547 secondes) de la durée du lot. Le pourcentage est encore plus élevé lorsque le nombre maximal de tâches simultanées est inférieur à 64.

Activation de la création de point de contrôle d’état asynchrone

Vous devez utiliser le magasin d’état RocksDB pour la réalisation d’un point de contrôle d’état asynchrone. Configurez les configurations suivantes :

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

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

Limitations et exigences relatives à la création de point de contrôle asynchrone

• Toute défaillance dans un point de contrôle asynchrone dans un ou plusieurs magasins échoue à la requête. En mode de création de point de contrôle synchrone, le point de contrôle est exécuté dans le cadre de la tâche et Spark tente d’exécuter la tâche à plusieurs reprises avant que la requête échoue. Ce mécanisme n’est pas présent avec la création de point de contrôle d’état asynchrone. Databricks conseille d’utiliser des travaux continus pour les nouvelles tentatives automatiques en cas d’échec de travail. Consultez Exécuter des travaux en continu.
• La création de point de contrôle asynchrone fonctionne de façon optimale quand les emplacements de magasin d’état ne sont pas modifiés entre les exécutions de micro-lots. Il est possible qu’un redimensionnement du cluster, en combinaison avec la création d’un point de contrôle d’état asynchrone, ne fonctionne pas correctement, car l’instance des magasins d’état peut être redistribuée à mesure que des nœuds sont ajoutés ou supprimés dans le cadre du redimensionnement du cluster.
• Le création de point de contrôle d’état asynchrone n’est prise en charge que dans l’implémentation du fournisseur de magasin d’état RocksDB. L’implémentation de magasin d’état en mémoire par défaut ne la prend pas en charge.