Asynchrone Zustandsprüfpunkte für zustandsbehaftete Abfragen

Asynchrone Zustandsprüfpunkte erhalten eine „Genau einmal“-Garantie für Streamingabfragen aufrecht, aber können beim strukturierten Streaming die Gesamtlatenz für einige zustandsbehaftete Workloads mit Engpässen bei Zustandsaktualisierungen verringern. Dies wird dadurch erreicht, dass mit der Verarbeitung des nächsten Mikrobatches begonnen wird, sobald die Berechnung des vorherigen Mikrobatches abgeschlossen ist, ohne dass auf den Abschluss der Zustandsprüfpunkte gewartet wird. In der folgenden Tabelle werden die Kompromisse für synchrone und asynchrone Checkpoints verglichen.

Im Folgenden sind Merkmale von Streamingaufträgen aufgeführt, die von asynchronen Zustandsprüfpunkten profitieren können:

• Der Auftrag weist zustandsbehaftete Vorgänge auf (z. B. Aggregation, flatMapGroupsWithState, mapGroupsWithState, Joins zwischen Streams).
• Die Latenz von Zustandsprüfpunkten trägt wesentlich zur Gesamtlatenz bei der Batchausführung bei. Diese Informationen finden Sie in den StreamingQueryProgress-Ereignissen. Diese Ereignisse sind auch in log4j-Protokollen im Spark-Treiber zu finden. Nachfolgend sehen Sie ein Beispiel für den Fortschritt einer Streamingabfrage und die Ermittlung der Auswirkungen des Zustandsprüfpunkts auf die Gesamtlatenz bei der Batchausführung.

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

  • Analyse der Latenz von Zustandsprüfpunkten beim oben angegebenen Abfragefortschrittsereignis

    • Die Batchdauer (durationMs.triggerDuration) beträgt ca. 547 Sekunden.
    • Die Latenz des Zustandsspeichercommits (stateOperations[0].commitTimeMs) beträgt ca. 3.186 Sekunden. Die Commit-Latenz wird für Aufgaben, die einen Zustandsspeicher enthalten, aggregiert. In diesem Fall gibt es 64 solcher Aufgaben (stateOperators[0].numShufflePartitions).
    • Jede Aufgabe, die den Zustandsoperator enthält, benötigte durchschnittlich 50 Sekunden (3.186/64) für den Prüfpunkte. Dies ist eine zusätzliche Latenz, die zur Batchdauer beiträgt. Wenn alle 64 Aufgaben gleichzeitig ausgeführt werden, macht der Prüfpunktschritt ungefähr 9 % (50 Sek./547 Sek.) der Batchdauer aus. Der Prozentsatz wird noch höher, wenn die maximale Anzahl gleichzeitiger Aufgaben kleiner als 64 ist.

Asynchrones Zustands-Checkpointing aktivieren

Sie müssen den RocksDB-basierten Zustandsspeicher für asynchrone Zustandsprüfpunkte verwenden. Legen Sie die folgenden Konfigurationen fest:

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

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

Einschränkungen und Anforderungen für asynchrone Prüfpunkte

• Jeder Fehler bei einem asynchronen Prüfpunkt in einem oder mehreren Speichern führt zu einem Fehler bei der Abfrage. Im synchronen Prüfpunktmodus wird der Prüfpunkt als Teil der Aufgabe ausgeführt, und Spark wiederholt die Aufgabe mehrmals, bevor die Abfrage fehlschlägt. Dieser Mechanismus ist bei asynchronen Zustandsprüfpunkten nicht vorhanden. Databricks empfiehlt die Nutzung fortlaufender Aufträge für automatische Wiederholungsversuche bei Auftragsfehlern. Weitere Informationen finden Sie unter Fortlaufendes Ausführen von Aufträgen.
• Asynchrone Prüfpunkterstellung funktioniert am besten, wenn die Speicherorte zwischen den Mikrobatchausführungen nicht geändert werden. Eine Größenanpassung des Clusters funktioniert in Kombination mit asynchronen Zustandsprüfpunkten möglicherweise nicht gut, da die Zustandsspeicherinstanz ggf. neu verteilt wird, wenn Knoten im Rahmen einer Änderung der Clustergröße hinzugefügt oder gelöscht werden.
• Asynchrone Zustandsprüfpunkte werden nur in der Implementierung des RocksDB-Zustandsspeicheranbieters unterstützt. Die Standardimplementierung des In-Memory-Zustandsspeichers unterstützt diese nicht.