Ausführen mehrerer strukturierter Streamingabfragen auf demselben Cluster

Viele Kunden führen mehrere Strukturierte Streaming-Abfragen auf demselben Azure Databricks Cluster aus. Obwohl dieses Muster unterstützt wird, empfiehlt Databricks, die Anzahl der Abfragen pro Cluster einzuschränken, um Skalierungsprobleme und Leistungsengpässe zu vermeiden. Bei der serverlosen Berechnung verwaltet Azure Databricks die Skalierung automatisch, sodass diese Überlegungen für Sie behandelt werden. Wenn Sie die klassische Berechnung verwenden, bei der Sie die Treiber- und Ausführungsgröße steuern, beschreibt diese Seite die wichtigsten Engpässe, um sie zu berücksichtigen und zu beheben.

Wann mehrere Abfragen im selben Cluster verwendet werden sollen

Das Ausführen mehrerer Streamingabfragen auf demselben Cluster reduziert die Infrastrukturkosten, insbesondere wenn Sie viele kleine Datenströme haben, für die nicht jeder dedizierte Compute erforderlich ist. Der Schlüsselkompromiss ist ein gemeinsamer Fehler: Wenn der Cluster fehlschlägt, schlägt jeder Datenstrom darauf fehl. Für unternehmenskritische Pipelines ist dieser Modus für gemeinsame Fehler häufig nicht akzeptabel.

Für Workloads, die kritische und nicht kritische Datenströme kombinieren, empfiehlt Databricks Folgendes:

• Weisen Sie jedem Datenstrom eine Priorität zu, basierend auf seinen geschäftlichen Auswirkungen.
• Platzieren Sie unternehmenskritische Datenströme auf dedizierten Clustern, auch bei höheren Kosten.
• Legen Sie Datenströme mit niedrigerer Priorität zusammen, um Rechenressourcen gemeinsam zu nutzen und Kosten zu senken.

Treiberdimensionierung

Der Treiber ist eine freigegebene Ressource. Mehrere Abfragen teilen die gleiche CPU, den Arbeitsspeicher, den DAG-Zeitplaner, den Aufgabenplaner und die treiberseitige UDF-Ausführung (z. B foreachBatch. ). Achten Sie bei der Ausführung vieler gleichzeitiger Streams auf diese spezifischen Engpässe, die über die standardmäßige CPU- und Speicherbereitstellung hinausgehen:

• Auto Loader-Mehraufwand: Wenn Ihre Streams Auto Loader verwenden, erhöhen Dateierkennung und Verzeichnisauflistung die Belastung des Treibers.
• Ressourcenbeschränkungen auf Betriebssystemebene (offene Dateien): Das Gleichzeitige Ausführen eines hohen Volumens dateibasierter Datenströme (z. B. auto FileStreamSource Loader) auf einem einzelnen Treiber kann die Grenzwerte für die offene Dateibeschreibung auf Benutzerebene ausschöpfen, was zu zufälligen Datenstromfehlern führen kann.
• Listener bus backpressure: Eine hohe Anzahl gleichzeitiger Streamingabfragen kann zu Rückdruck auf dem Bus der einzelnen Spark-Sitzung StreamingQueryListener führen. Alle Ereignisse (einschließlich onQueryIdle) werden an diesen einzelnen Bus gesendet, und ein großer Ereignis-Backlog kann asynchrone onQueryProgress Handler erheblich verzögern und die Clusterstabilität beeinträchtigen.
• Kostspielige Vorgänge auf dem Treiber: Vermeiden Sie Aufrufe von collect() oder anderen kostspieligen DataFrame-Operationen auf dem Treiber, sofern dies nicht unbedingt erforderlich ist, da sonst große Ergebnismengen materialisiert werden und Out-of-Memory-Fehler (OOM) auftreten können.

Beheben von Treiberkonflikten

Wenn bei Ihnen Treiberabstürze aufgrund von OOM-Problemen oder Konflikten auftreten:

1. Überwachen sie die Treibermetriken in der Spark-Benutzeroberfläche. Wenn eine hohe CPU-, Arbeitsspeicher- oder Datenträgerauslastung angezeigt wird, passen Sie die Größe des Treibers in den Cluster-Computeeinstellungen an.
2. Wenn Probleme weiterhin bestehen, stellen Sie sicher, dass der Code keine speicherintensiven Vorgänge oder UDFs auf dem Treiber ausführt.
3. Wenn Sie den Treiber nicht vertikal skalieren können, empfiehlt Databricks dringend, Ihre Aufträge auf mehrere Cluster aufzuteilen, um diese Engpässe bei der Skalierung gemeinsam genutzter Knoten zu umgehen.

Executor-Dimensionierung

Bei mehreren Abfragen, die auf demselben Cluster ausgeführt werden, geben alle Abfragen Aufgabenplätze für die Executoren frei. Phasen einer Abfrage können verfügbare Slots belegen, was zu Verzögerungen und zum Aushungern anderer Abfragen führt. Spark verwendet eine 1:1-Zuordnung zwischen Aufgabenplätzen und verfügbaren Kernen. Stellen Sie sicher, dass genügend Kerne verfügbar sind, wenn Abfragen gleichzeitig ausgeführt werden müssen.

Im Allgemeinen können Executoren mehr arbeitsspeicherintensive Vorgänge ausführen als der Treiberknoten. Passen Sie bei Bedarf die Executor-JVM sowie die Parameter für die Speicherzuweisung außerhalb des Heaps an, um Ihre Anwendungslast zu bewältigen. Stellen Sie sicher, dass die Executor-Knoten in Bezug auf CPU, Arbeitsspeicher und Festplattenspeicher angemessen dimensioniert sind und bei Bedarf vertikal skaliert werden. Wenn die vertikale Skalierung nicht möglich ist, sollten Sie dem Cluster zusätzliche Arbeitsknoten hinzufügen.

Scheduler-Pools verwenden

Sie können Schedulerpools so konfigurieren, dass Abfragen Rechenkapazität zugewiesen werden, wenn mehrere Streamingabfragen aus demselben Quellcode ausgeführt werden.

Standardmäßig werden alle in einem Notebook gestarteten Abfragen im gleichen fairen Planungspool ausgeführt. Apache Spark-Jobs, die durch Trigger aller Streaming-Abfragen in einem Notebook generiert werden, werden nacheinander in der Reihenfolge „First In, First Out“ (FIFO) ausgeführt. Dies kann zu unnötigen Verzögerungen in den Abfragen führen, da sie die Clusterressourcen nicht effizient freigeben.

Mit Scheduler-Pools können Sie deklarieren, welche strukturierten Streaming-Abfragen Computeressourcen gemeinsam nutzen.

Im folgenden Beispiel wird query1 einem dedizierten Pool zugewiesen, während query2 und query3 einen gemeinsamen Planerpool nutzen.

:::note Serverless-Kompatibilität

Databricks empfiehlt, von spark.sparkContext abzurücken, da spark.sparkContext nicht mit der serverlosen Compute-Architektur von Databricks kompatibel ist. Verwenden Sie spark (SparkSession) stattdessen direkt. Planerpools sind ein klassisches Computekonzept; auf serverlosen Servern verwaltet Databricks die Skalierung und Ressourcenzuordnung automatisch.

:::

# Run streaming query1 in scheduler pool1
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool1")
df.writeStream.queryName("query1").toTable("table1")

# Run streaming query2 in scheduler pool2
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
df.writeStream.queryName("query2").toTable("table2")

# Run streaming query3 in scheduler pool2
spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
df.writeStream.queryName("query3").toTable("table3")

Weitere Informationen zu den Fair-Scheduler-Pools finden Sie in der Apache Spark Fair Scheduler-Dokumentation.

Überlegungen zu zustandsbehafteten Abfragen

Beachten Sie bei zustandsbehafteten Abfragen, die auf demselben Cluster ausgeführt werden, Folgendes:

• Verwenden Sie RocksDB als State Store-Anbieter , um OOM-Probleme und GC-Pausen zu vermeiden. RocksDB ist der Standardstatusspeicheranbieter in Databricks Runtime 17.3 und höher. Weitere Informationen finden Sie unter Konfigurieren des RocksDB-Statusspeichers auf Azure Databricks.
• Optimieren Sie die Shuffle-Partitionen für Ihre Anwendungsanforderungen. Bei zustandsbehafteten Phasen plant Spark Vorgänge proportional zur Anzahl der Shuffle-Partitionen.
• Begrenzen Sie die Speichernutzung von RocksDB pro Knoten, um OOM-Fehler durch Off-Heap-Speichernutzung zu vermeiden. Dies wird automatisch in Databricks Runtime 17.3 und höher behandelt, erfordert jedoch eine manuelle Konfiguration in früheren Versionen. Siehe Cap RocksDB-Speicherauslastung.
• Vermeiden Sie, zu viele Partitionen auf demselben Executor-Knoten zu platzieren. Wartungsvorgänge im Zustandsspeicher, einschließlich Snapshot-Upload und Bereinigung, werden auf Knotenbasis ausgeführt. Wenn einem Executor-Knoten zu viele Partitionen zugewiesen werden, kann dies zu Wartungsengpässen und längeren Wiederherstellungszeiten führen, da weniger vollständige Snapshots verfügbar sind.