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 les requêtes de diffusion en continu, mais elle peut augmenter la latence globale pour certaines charges de travail Structured Streaming avec état, qui sont limitées par les 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 des tâches de streaming susceptibles de bénéficier du point de contrôle d’état asynchrone :

• Le travail contient une ou plusieurs opérations avec état (par exemple, agrégation, flatMapGroupsWithState, mapGroupsWithState, jointures de flux à flux)
• La latence du point de contrôle est l’un des facteurs majeurs de la latence d’exécution globale d’un processus en lots. Ces informations se trouvent dans les événements StreamingQueryProgress. Ces événements se trouvent également dans les journaux de bord log4j du pilote Spark. Voici un exemple de progression d'une requête en streaming et comment trouver l’impact du point de contrôle d’état sur la latence globale d'exécution par 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 la requête susmentionnée

    • 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 à travers 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 effectuer le point de contrôle. Il s'agit d'une latence supplémentaire qui contribue à la durée du traitement par lots. 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 du 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 d'un point de contrôle asynchrone dans un ou plusieurs stockages entraîne l'échec de la requête. En mode de point de contrôle synchrone, le point de contrôle est exécuté dans le cadre de la tâche, et Spark réessaie d’exécuter la tâche plusieurs fois avant que la requête n’échoue. Ce mécanisme n’est pas présent avec le point de contrôle d’état de manière asynchrone. Databricks conseille d’utiliser des jobs continus pour les nouvelles tentatives automatiques en cas d’échec. Consultez Exécuter des travaux en continu.
• La point de contrôle asynchrone fonctionne mieux lorsque les emplacements de stockage 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 de l'état asynchrone, ne fonctionne pas correctement, car l'instance de stockage d’état pourrait être redistribuée à mesure que des nœuds sont ajoutés ou supprimés dans le cadre du redimensionnement du cluster.
• La création de point de contrôle d’état asynchrone n’est prise en charge que dans l’implémentation du fournisseur de stockage d’état RocksDB. L’implémentation de magasin d’état en mémoire par défaut ne la prend pas en charge.