Produktionsüberlegungen für strukturiertes Streaming

Führe Produktions-Structured Streaming-Workloads wie geplante Lakeflow-Jobs oder Lakeflow-Pipelines aus. Siehe Lakeflow Jobs und Spark deklarative Pipelines auf Lakeflow.

Nutzen Sie Lakeflow-Pipelines, um Orchestrierung, Checkpoints und Wiederherstellung zu verwalten. Verwenden Sie serverlose Pipelines, um Rechenressourcen zu verwalten und sie automatisch zu skalieren. Siehe Konfigurieren einer serverlosen Pipeline.

Konfigurieren Sie Streaming-Workloads, die als Lakeflow-Jobs laufen, wie folgt:

  • Entfernen Sie unnötigen Code aus Notebooks, der Ergebnisse zurückgeben würde, z. B. display und count.
  • Führen Sie keine strukturierten Streaming-Workloads mit allzweckbasierter Berechnung aus. Planen Sie Datenströme immer als Lakeflow-Einzelvorgänge mit Job Compute.
  • Planen von Lakeflow-Aufträgen mithilfe des Continuous Modus. Dies bezieht sich auf das Planungsfeature für Azure Databricks-Einzelvorgänge, nicht auf das Intervall für strukturiertes Streaming Triggerintervall.
  • Aktiviere keine automatische Skalierung für Structured Streaming-Jobs. Autoscaling hat Einschränkungen, wenn es darum geht, die Clustergröße für Streaming-Workloads zu reduzieren.

Einige Workloads profitieren von folgendem:

Für Notebooks und Jobs auf serverlosem Compute werden nur Trigger.AvailableNow() und Trigger.Once() unterstützt. Databricks empfiehlt Trigger.AvailableNow(). Siehe Streaming-Einschränkungen.

Verwenden Sie für kontinuierliches Streaming auf serverlosem Computing Lakeflow-Pipelines im kontinuierlichen Modus. Siehe Ausgelöste vs. Continuous Pipeline-Modus.

Reduzieren Sie die Latenz beim operativen Streaming

Operative Streaming-Workloads nehmen Daten nahezu in Echtzeit ein, transformieren und reagieren darauf. Typische Beispiele sind Betrugserkennung, Anomalieerkennung, Personalisierung sowie Echtzeitüberwachung und -benachrichtigungen, wobei verzögerte Verarbeitung direkt die Geschäftsergebnisse beeinflusst. Niedrige Latenz für diese Workloads bedeutet typischerweise Dutzende bis Hunderte von Millisekunden, wobei viele Teams Service Level Agreements (SLAs) im Sekundenbereich festlegen, um Schwankungen in höheren Perzentilen zu berücksichtigen.

Wählen Sie einen Verarbeitungsmodus basierend auf den Latenzanforderungen Ihrer Workload:

  • Nutzen Sie den Echtzeitmodus (RTM) für operative Workloads mit Latenzen unter einer Sekunde. RTM ist bis zu 100-mal schneller als der Micro-Batch-Modus, mit einer End-to-End-Latenz von nur fünf Millisekunden.
  • Verwenden Sie den Mikro-Batch-Modus für analytische Pipelines und ETL-Workloads mit Latenzanforderungen, die in Sekunden oder Minuten gemessen werden.

Siehe Echtzeit-Modus-Konzepte.

Wenn der Echtzeitmodus nicht zu Ihrer Arbeitslast passt, reduzieren die folgenden Best Practices die Latenz für Micro-Batch Structured Streaming:

  • Ausgabemodus: Nutze den Update-Modus, in dem deine Abfrageoperatoren und Sinks ihn unterstützen. Der Update-Modus gibt nach jeder Auslösung aktualisierte Zeilen aus und gibt sie weiterhin aktualisiert aus, bis das Watermark abläuft. Stellen Sie daher sicher, dass Ihr nachgelagerter Sink idempotent ist, damit aktualisierte Ergebnisse verarbeitet werden können. Nutze den Append-Modus für Workloads, die vom Updatemodus nicht unterstützt werden, wie z. B. Stream-Stream-Joins, oder wenn du verspätet eintreffende Daten verwerfen kannst. Benutze nicht den Komplettmodus wegen niedriger Latenz. Siehe Auswählen eines Ausgabemodus für strukturiertes Streaming.
  • Trigger: Verwenden Sie einen processingTime Trigger mit einem Intervall 0 , der die nächste Mikrocharge startet, sobald die vorherige abgeschlossen ist und neue Daten verfügbar sind. Dies bietet die niedrigste Micro-Batch-Latenz, erhöht aber die Kosten für Cloud-Speicher-APIs. Verwenden Sie AvailableNow, Once oder Continuous nicht für Produktivworkloads. Weitere Informationen finden Sie unter Konfigurieren von Triggerintervallen für strukturiertes Streaming.
  • Wasserzeichen: Legen Sie das Wasserzeichen so lang fest, dass auch die verspätet eintreffenden Daten berücksichtigt werden, die Ihre Workload auf keinen Fall verwerfen darf. Das Wasserzeichen steuert, wie lange die Abfrage nicht in der Reihenfolge eintreffende Ereigniszeitdaten akzeptiert, bevor sie diese verwirft und der Entfernungsstatus auftritt; ein zu kurzes Wasserzeichen verwirft daher stillschweigend gültige verspätet eintreffende Datensätze. Innerhalb dieser Einschränkung senkt ein kürzeres Wasserzeichen die Latenz und behält weniger Zustand, während ein längeres Wasserzeichen mehr späte Daten toleriert – auf Kosten von Latenz und Zustand. Ein kleines Vielfaches deiner Latenz-SLA, zum Beispiel 2x, ist ein vernünftiger Ausgangspunkt für das Tuning. Siehe Anwenden von Wasserzeichen zum Steuern von Schwellenwerten für die Datenverarbeitung.
  • Quellen und Sinks: Lesen Sie von Quellen mit niedriger Latenz wie Nachrichtenbussen (Apache Kafka, Amazon Kinesis, Apache Pulsar oder Google Cloud Pub/Sub) oder ändern Sie Datenfeeds aus Delta Lake und Apache Iceberg Tabellen. In latenzarme Sinks mit hohem Durchsatz wie Nachrichtenbusse, operative Datenbanken oder foreach-Sinks schreiben. Entwerfen Sie Sink-Operationen so, dass sie idempotent sind, damit nachgelagerte Verbraucher doppelte und spät eintreffende Daten verarbeiten.
  • Zustand und Prüfpunkte: Für zustandsbehaftete Abfragen verwenden Sie den RocksDB-Zustandsspeicher, der sowohl für Changelog-Prüfpunkte als auch für asynchrone Zustandsprüfpunkte erforderlich ist. Aktivieren Sie Changelog-Checkpointing, um nur inkrementelle Zustandsänderungen beizubehalten. Wenn das Schreiben von Status-Checkpoints der Engpass für die Dauer Ihres Batches ist, aktivieren Sie asynchrones Status-Checkpointing, um das Schreiben der Checkpoints mit dem nächsten Mikrobatch zu überlappen, nachdem Sie die Einschränkungen bei der Fehlerwiederherstellung und der Größenanpassung des Clusters geprüft haben. Geben Sie jeder Abfrage ein eigenes Checkpoint-Verzeichnis in einem dauerhaften Cloud-Speicher. Siehe RocksDB-Statusspeicher in Azure Databricks konfigurieren, Asynchrone Status-Checkpoint-Erstellung für zustandsbehaftete Abfragen und Structured-Streaming-Checkpoints.
  • Offset-Management: Um die Latenz durch Offset-Checkpoints in kontinuierlichen Streams zu reduzieren, aktivieren Sie asynchrone Fortschrittsverfolgung, die die Offset- und Commit-Logs aktualisiert, ohne die Datenverarbeitung zu blockieren. Es ist nicht mit den Triggern AvailableNow oder Once kompatibel. Siehe asynchrone Statusnachverfolgung.
  • Speichersprünge: Halten Sie die Verarbeitung nach Möglichkeit in einer einzigen Streaming-Pipeline. Das Aufteilen der Logik auf mehrere Jobs oder Pipelines fügt Speicher-Hops hinzu, die die Latenz erhöhen.

Streaming-Workloads darauf auslegen, Ausfälle zu erwarten

Databricks empfiehlt, Streamingaufträge immer so zu konfigurieren, dass beim Fehler automatisch neu gestartet wird. Einige Funktionen, einschließlich der Schemaentwicklung, erfordern, dass strukturierte Streaming-Workloads automatisch erneut versuchen. Weitere Informationen finden Sie unter Konfigurieren von strukturierten Streamingaufträgen zum Neustart von Streamingabfragen bei Fehlern.

Einige Vorgänge wie foreachBatch bieten eine Garantie vom Typ „Mindesten einmal“ statt „Genau einmal“. Stellen Sie für diese Vorgänge sicher, dass Ihre Verarbeitungspipeline idempotent ist. Siehe Verwenden von foreachBatch zum Schreiben in beliebige Datensenken.

Hinweis

Wenn eine Abfrage neu gestartet wird, wird der bei der letzten Ausführung geplante Mikrobatch verarbeitet. Wenn ein Auftrag aufgrund von ungenügendem Arbeitsspeicher fehlgeschlagen ist oder Sie einen Auftrag aufgrund eines übergroßen Mikrobatches manuell abgebrochen haben, müssen Sie das Compute möglicherweise skalieren, um den Mikrobatch erfolgreich zu verarbeiten.

Wenn Sie Konfigurationen zwischen Ausführungen ändern, gelten die betreffenden Konfigurationen für den ersten geplanten neuen Batch. Siehe "Wiederherstellen nach Änderungen in einer strukturierten Streamingabfrage".

Beim erneuter Ausführung eines Einzelvorgangs

Sie können mehrere Vorgänge als Teil eines Azure Databricks Auftrags planen. Wenn Sie einen Auftrag mit dem Trigger „Fortlaufend“ konfigurieren, können Sie keine Abhängigkeiten zwischen Aufgaben festlegen.

Für die Planung mehrerer Streams in einem einzigen Auftrag stehen Ihnen die folgenden Vorgehensweisen zur Verfügung:

  • Mehrere Aufgaben: Definieren Sie einen Auftrag mit mehreren Aufgaben, die Streaming-Workloads mithilfe eines fortlaufenden Auslösers ausführen.
  • Mehrere Abfragen: Definieren mehrerer Streamingabfragen im Quellcode für eine einzelne Aufgabe.

Diese Strategien lassen sich auch kombinieren. Im folgenden Diagramm werden diese beiden Vorgehensweisen miteinander verglichen.

Strategie Mehrere Aufgaben Mehrere Abfragen
Wie wird Compute aufgeteilt? Databricks empfiehlt die Bereitstellung von Jobs Compute, das für die einzelnen Streamingaufgaben entsprechend dimensioniert ist. Sie können optional die Rechenleistung auf Aufgaben verteilen. Alle Abfragen teilen sich dasselbe Compute. Sie können optional Abfragen den Scheduler-Pools zuweisen.
Wie werden Wiederholungsversuche gehandhabt? Ein fehlgeschlagener Task wird nur dann automatisch wiederholt, wenn Task retry mode auf On failure gesetzt ist. Ansonsten müssen alle Vorgänge fehlschlagen, bevor der Auftrag erneut ausgeführt wird. Die Aufgabe wird wiederholt, wenn eine der Abfragen fehlschlägt.

Warning

Wenn eine Streaming-Aufgabe in einem Multitask-Continuous Job fehlschlägt, während andere Aufgaben weiterlaufen, startet der Job erst einen neuen Lauf, wenn alle anderen Aufgaben ebenfalls ausfallen. Der ausgefallene Stream kann ausgefallen bleiben, solange der Job funktionsfähig wirkt. Um dieses Risiko zu verringern, setzen Sie den Aufgaben-Wiederholungsmodus auf Bei Fehlschlag und Benachrichtigungen konfigurieren für Aufgabenfehler. Alternativ kannst du jeden kritischen Stream als eigenen Job ausführen. Siehe Fehlerbehandlung für kontinuierliche Einzelvorgänge.

Weitere Informationen zum Arbeiten mit mehreren Aufgaben oder Abfragen finden Sie unter Ausführen mehrerer strukturierter Streamingabfragen auf demselben Cluster.

Konfigurieren von strukturierten Streaming-Aufträgen zum Neustarten von Streaming-Abfragen bei einem Fehler

Databricks empfiehlt, alle Streamingworkloads mithilfe des fortlaufenden Triggers zu konfigurieren. Weitere Informationen finden Sie unter Fortlaufendes Ausführen von Aufträgen.

Der fortlaufende Auslöser weist standardmäßig das folgende Verhalten auf:

  • Er verhindert, dass der Auftrag mehr als einmal gleichzeitig ausgeführt wird.
  • Er startet eine neue Ausführung, wenn eine vorherige Ausführung fehlschlägt.
  • Er nutzt ein exponentielles Backoff-Verfahren für Wiederholungsversuche.

Databricks empfiehlt, bei der Planung von Workflows stets Jobs Compute zu verwenden statt All-Purpose Compute. Beim Fehlschlagen und Wiederholen von Aufträgen werden neue Computeressourcen bereitgestellt.

Hinweis

Databricks empfiehlt, nicht zu verwenden streamingQuery.awaitTermination() oder spark.streams.awaitAnyTermination(). Siehe Wann awaitTermination() zu verwenden ist.

Wann verwendet werden soll awaitTermination()

streamingQuery.awaitTermination() und spark.streams.awaitAnyTermination() blockieren den aktuellen Thread, bis eine Streaming-Abfrage beendet wird. Ob diese Funktionen verwendet werden sollen, hängt von Ihrer Ausführungsumgebung ab.

Verwenden Sie für Lakeflow Jobs nicht streamingQuery.awaitTermination() oder spark.streams.awaitAnyTermination(). Diese Funktionen sind nicht erforderlich, da der Jobs-Dienst automatisch verhindert, dass ein Durchlauf abgeschlossen wird, wenn eine Streamingabfrage aktiv ist. Beide Funktionen verhindern, dass Notebookzellen ihren Lauf abschließen, und verhindern, dass der Einzelvorgangsdienst die Streamingabfrage nachverfolgt, was zu Unterbrechungen bei Backlogmetriken und Einzelvorgangsbenachrichtigungen führt.

Verwenden Sie awaitTermination() in den folgenden Fällen:

Anwendungsfall Verhalten
Interaktive Notebooks für den All-Purpose Compute awaitTermination() hält die Zelle aktiv, ermöglicht es Ihnen, den Abfragezustand zu beobachten und stellt sicher, dass Fehler in der Notizbuchausgabe angezeigt werden.
Lokale und Entwicklungsumgebungen Wenn Sie ein Spark-Programm lokal ausführen, wird der Prozess beendet, wenn der Hauptthread abgeschlossen ist. Rufen Sie auf awaitTermination() , um das Programm lebendig zu halten, bis die Streamingabfrage abgeschlossen ist oder fehlschlägt.
Fehlerverteilung an den Treiber Ohne awaitTermination() könnte ein Streamingabfragefehler in einem Nichtauftragskontext möglicherweise nicht an den aufrufenden Thread weitergegeben werden. Die Abfrage kann im Hintergrund fehlschlagen, wodurch Fehler schwieriger zu erkennen und zu diagnostizieren sind. Beim Aufruf von awaitTermination() wird die Abfrageausnahme im Treiber erneut ausgelöst.
  • Zuletzt aktualisiert um