Ankündigung der On-Demand-Zustandsrepartitionierung für Apache Spark™ Structured Streaming auf Databricks

Jeder, der zustandsbehaftete Apache Spark™ Structured Streaming-Abfragen in der Produktion ausführt, stößt irgendwann an dieselbe unangenehme Grenze.

Sie haben die Abfrage vor Monaten gestartet. Damals war das Datenvolumen noch überschaubar, also haben Sie den Standardwert von 200 Shuffle-Partitionen übernommen und weitergemacht. Die Pipeline lief reibungslos. Dann wuchs das Geschäft, der Traffic verdreifachte sich und der State Store blähte sich auf. Plötzlich haben diese 200 Partitionen nicht mehr die richtige Größe. Einige Partitionen sind ungleichmäßig verteilt und laufen heiß, der Cluster stößt an seine Grenzen und jeder Micro-Batch dauert länger als er sollte.

Also tun Sie das Naheliegende: Sie erhöhen spark.sql.shuffle.partitions und starten die Abfrage neu. Nichts ändert sich.

Die Abfrage ignoriert Ihren neuen Wert stillschweigend, da die Anzahl der Partitionen beim ersten Start des Streams im Checkpoint festgeschrieben wurde. Bisher bestand die einzige Möglichkeit, eine neue Anzahl anzuwenden, darin, den vorhandenen Checkpoint zu verwerfen und von vorne zu beginnen. Bei einer zustandsbehafteten Abfrage bedeutet dies jedoch den Verlust des gesamten mühsam aufgebauten Zustands, den Sie sorgfältig gepflegt haben. Bei einem Betrugserkennungsmodell, das Millionen von Konten verfolgt, oder einem Sessionisierungs-Job, der Zeitfenster über mehrere Tage hinweg speichert, ist „von vorne beginnen“ kein Satz, den man bei einer Fehleranalyse in der Produktion hören möchte.

Zustandsbehaftete Repartitionierung auf Abruf (Public Preview), verfügbar in Databricks Runtime 18 und höher, beseitigt diese Barriere. Sie können jetzt die Anzahl der Partitionen für eine zustandsbehaftete Streaming-Abfrage anpassen und Ihren Checkpoint-Zustand beibehalten.  

Dies gilt für jede zustandsbehaftete Streaming-Abfrage, unabhängig davon, ob Sie Aggregationen, Stream-Stream-Joins, Deduplizierung, Sessionisierung oder transformWithState ausführen, und für jeden Workload, von der Betrugserkennung bis hin zur Echtzeitüberwachung.

Für Early Adopter wie Coveo führte die Möglichkeit, ihre Streaming-Infrastruktur bei Bedarf optimal zu dimensionieren, sofort zu erheblichen Betriebseinsparungen. 

„Bei Coveo betreiben wir umfangreiche zustandsbehaftete Streaming-Pipelines, bei denen das Datenvolumen im Laufe der Zeit stark schwankt. Mit Databricks und der Funktion zur zustandsbehafteten Repartitionierung konnten wir unsere damit verbundenen Amazon S3 API-Kosten um 40 % senken.  Zuvor erforderte jede Skalierungsentscheidung einen Kompromiss: Entweder mussten wir überdimensionieren oder neu aus neuen Checkpoints aufbauen, was die Kosten für Speicher-APIs fast auf das Niveau der Rechenkosten trieb. Jetzt können wir bei Nachfrageänderungen frei skalieren, ohne den bestehenden Zustand zu beeinträchtigen oder kostspielige Checkpoint-Migrationen auszulösen.“  —Alexis Chicoine, Senior Software Developer, Coveo

Ein Blick hinter die Kulissen: Warum waren Zustandspartitionen gesperrt?

Um zu verstehen, warum die Ergebnisse von Coveo einen bedeutenden Fortschritt für Structured Streaming darstellen, müssen wir uns ansehen, warum die Partitionsanzahl überhaupt jemals eingefroren wurde.

Eine zustandsbehaftete Streaming-Abfrage speichert ihren Zustand in einem State Store, und dieser Zustand ist physisch partitioniert. Jeder Schlüssel in Ihrem Stream – z. B. eine Benutzer-ID, eine Kontonummer oder ein Zeitfenster – wird einer bestimmten Partition zugewiesen (gehasht), und die Daten für jede Partition werden in einer eigenen, separaten RocksDB-Instanz innerhalb des Checkpoints gespeichert. Die Anzahl der Partitionen bestimmt das Layout des gesamten State Stores auf dem Datenträger.

Wenn Sie die Partitionsanzahl einfach zwischen Neustarts ändern würden, würde das Hashing nicht mehr übereinstimmen. Ein Schlüssel, der zuvor in einer Partition lag (z. B. Partition 47), würde nun einer anderen Partition zugewiesen (z. B. Partition 12), aber sein akkumulierter Zustand befindet sich immer noch in den Dateien der ursprünglichen Partition. Die Abfrage würde praktisch den Überblick über ihren eigenen Speicher verlieren. Um genau diese Art von unbemerkter Datenbeschädigung zu verhindern, hat Structured Streaming die Partitionsanzahl bei der Erstellung des Checkpoints gesperrt und spätere Änderungen an spark.sql.shuffle.partitions ignoriert.

Sicher, aber unflexibel. Die beiden Nachteile dabei waren:

1. Sie konnten keine Optimierungen vornehmen. Wenn sich herausstellte, dass 200 Partitionen die falsche Wahl waren, waren Sie für die gesamte Lebensdauer des Checkpoints daran gebunden.
2. Sie konnten nicht mit dem Workload skalieren. Wenn das Datenvolumen stieg oder sank, konnte Ihre Partitionsanzahl nicht Schritt halten.

Die zustandsbehaftete Repartitionierung auf Abruf löst beide Probleme, indem sie genau das tut, was das alte Design verweigert hat – dies jedoch auf sichere Weise, indem sie den Zustand physisch neu verteilt, um ihn an die neue Partitionsanzahl anzupassen.

Was Sie für den Einstieg benötigen

Die Voraussetzungen sind überschaubar:

• Databricks Runtime 18 oder höher.
• Der RocksDB-State-Store-Provider. In DBR 17.3 und höher ist RocksDB der Standard. Neue Abfragen, die in diesen Versionen erstellt werden, verwenden RocksDB, sofern dies nicht explizit geändert wird.  Wenn Sie dies überprüfen oder explizit festlegen möchten, lesen Sie RocksDB-State-Store auf Databricks konfigurieren.

Das ist schon die ganze Liste der Voraussetzungen. Wenn Sie DBR 18 mit dem Standard-State-Store verwenden, haben Sie bereits alles, was Sie brauchen.

Ändern der Partitionsanzahl

Der Mechanismus ist einfach und nutzt ein Muster, das jeder Streaming-Entwickler bereits kennt: Stoppen, neu konfigurieren, neu starten.

Anstelle von spark.sql.shuffle.partitions legen Sie eine dedizierte Konfiguration fest,  spark.sql.streaming.stateStore.partitions, und starten die Abfrage neu:

Das entscheidende Detail ist die neue Konfiguration selbst. Bei zustandsbehafteten Abfragen hat spark.sql.streaming.stateStore.partitions Vorrang vor spark.sql.shuffle.partitions. Dadurch wird die Änderung im Gegensatz zum alten Ansatz dauerhaft übernommen.

Wenn die Abfrage neu startet, wird die normale Verarbeitung nicht sofort fortgesetzt. Zuerst wird der letzte geplante Micro-Batch abgeschlossen, falls noch einer aussteht. Anschließend wird ein einmaliger Repartitionierungsvorgang durchgeführt: Die Zustandsdaten werden physisch auf die neue Anzahl von Partitionen verteilt, wobei die Schlüssel in ihre neuen, korrekten Partitionen gehasht werden, sodass nichts verloren geht oder falsch zugeordnet wird. Sobald diese Neuverteilung abgeschlossen ist, setzt die Abfrage die Verarbeitung wie gewohnt fort und verwendet nun die von Ihnen angeforderte Partitionsanzahl.

Dieser Repartitionierungsschritt ist das Herzstück der Funktion. Er macht den Unterschied aus zwischen „Wir haben eine Zahl geändert“ und „Wir haben Ihren Zustand sicher in ein neues Layout verschoben“.

Überwachung des Repartitionierungsvorgangs

Da die Repartitionierung ein tatsächlicher Vorgang ist, dessen Laufzeit proportional zur Menge des Zustands ist, möchten Sie sicherlich Einblick darin haben. Structured Streaming stellt dies über seine standardmäßige Fortschrittsberichterstattung bereit.

Nach Abschluss des nächsten Micro-Batches enthalten die StreamingQueryProgress-Ereignisse die Dauer des Repartitionierungsvorgangs. Suchen Sie in den durationMs-Metriken des Ereignisses nach dem Feld controlBatch.REPARTITION, das die Dauer der Repartitionierung in Millisekunden angibt.

Ein größerer Zustand führt zu einer längeren Repartitionierung, aber wir gehen davon aus, dass dies bei den meisten Workloads nur wenige Sekunden dauert. Bei großen Jobs lohnt es sich daher, diese Metrik zu erfassen, um die Dauer zu verstehen. Weitere Informationen zum Lesen dieser Ereignisse finden Sie unter Überwachung von Structured Streaming-Abfragen auf Databricks.

Beispiel: Herunterskalieren einer Abfrage

Machen wir dies an einer einfachen Aggregation konkret – einer Zählung von Ereignissen nach ID in einem Tumbling-Window (rollenden Zeitfenster). Wir starten mit dem Standardwert von 200 Partitionen, stellen fest, dass dies mehr ist, als dieser Workload benötigt, und skalieren auf 100 herunter.

Zuerst die Abfrage, wie sie heute mit der Standard-Partitionsanzahl ausgeführt wird:

Nun haben wir diesen Stream eine Weile beobachtet und sind zu dem Schluss gekommen, dass 200 Partitionen übertrieben sind. Wir zahlen Koordinations-Overhead für eine Parallelität, die wir nicht benötigen. Wir stoppen die Abfrage, legen die neue Partitionsanzahl fest und starten sie mit denselben Optionen und demselben Checkpoint neu:

Wenn die neu gestartete Abfrage hochfährt, schließt sie den letzten geplanten Micro-Batch ab (falls noch einer aussteht), führt die Repartitionierung aus, um den Zustand von 200 Partitionen auf 100 umzuverteilen, und zählt dann weiter, wobei jedes Zeitfenster und jede laufende Summe vollständig erhalten bleiben. Dasselbe Verfahren funktioniert auch umgekehrt: Um bei einer höheren Last hochzuskalieren, legen Sie einfach eine größere Zahl fest.

Derselbe Ansatz gilt auch für Spark Declarative Pipelines (SDP). In der Dokumentation finden Sie ein SDP-Beispiel mit einer vollständigen Anleitung.

Wann Sie die State-Repartitionierung verwenden sollten

Die On-Demand-State-Repartitionierung ist eher ein Tool zur Optimierung und Skalierung als ein Routinevorgang. Sie erweist sich in einigen Schlüsselsituationen als wertvoll: 

• Passende Dimensionierung nach dem Start. Sie haben die Pipeline am ersten Tag mit den standardmäßigen 200 Partitionen gestartet, weil der Stream klein war und sich eine Feinabstimmung nicht lohnte. Sechs Monate später ist diese Zahl fest in einem Checkpoint verankert, den Sie sich nicht leisten können zu verlieren, und sie reicht nicht mehr aus. Beispiel: Ein Stream zur Betrugsbewertung (Fraud Scoring), der in einer einzelnen Pilotregion gestartet wurde, deckt nun alle Märkte ab, und bei 200 Partitionen enthält jede einzelne viel zu viel State. Mit der On-Demand-Repartitionierung können Sie die Anzahl der Partitionen an Ihre aktuelle Last anpassen, ohne Ihren vorhandenen Checkpoint zu verlieren.
• Wechselnde Workloads. Sie haben den Stream für die Spitzenlast dimensioniert. Beispiel: Eine Ad-Bidding-Pipeline läuft tagsüber heiß und wird nachts ruhig. Ein für die Tagesspitze optimierter Wert führt also dazu, dass die meisten Partitionen um 3 Uhr morgens im Leerlauf sind. Mit der On-Demand-Repartitionierung skalieren Sie vor der Hauptverkehrszeit hoch und danach wieder herunter, sodass die Partitionierung der tatsächlichen Last und nicht dem Worst-Case-Szenario folgt.
• Backfilling historischer Daten:  Backfill- und Steady-State-Verarbeitung erfordern unterschiedliche Partitionszahlen. Bisher mussten Sie sich für die gesamte Lebensdauer des Checkpoints für eine entscheiden. Beispiel: Die erneute Verarbeitung von zwei Jahren Historie erfordert eine hohe Anzahl, um die Arbeit zu verteilen und schnell fertig zu werden. Dieselbe Anzahl ist jedoch verschwenderisch, sobald Sie wieder beim normalen Steady-State-Traffic sind. Mit der On-Demand-Repartitionierung können Sie für das Backfilling hochskalieren und nach dem Aufholen wieder auf die Steady-State-Größe herunterskalieren – und das alles, ohne den Checkpoint und den State zu verlieren.
• Performance-Tuning. Die Anzahl der Partitionen beeinflusst die Parallelität, die State-Größe und den Shuffle-Overhead, und der optimale Wert ist schwer vorherzusagen. Beispiel: Sie denken vielleicht, dass 200 zu wenig sind und 400 die Microbatch-Latenz verringern würden. Früher erforderte das Testen jedoch den Neuaufbau des States und die erneute Verarbeitung von Daten, was Ressourcen verschwendete. Mit der On-Demand-Repartitionierung können Sie die Anzahl anhand Ihres aktiven Checkpoints anpassen, die Dauer von controlBatch.REPARTITION und Microbatches überwachen und Entscheidungen auf der Grundlage von Messungen statt Vermutungen treffen.

Da jede Änderung einen Stopp und Neustart mit einer einmaligen Repartitionierungspause erfordert, sollten Sie dies als bewusste Wartungsmaßnahme behandeln. Planen Sie die Größenänderung für ein Zeitfenster, in dem eine kurze Verarbeitungspause akzeptabel ist, beobachten Sie controlBatch.REPARTITION, um zu bestätigen, wie lange es gedauert hat, und lassen Sie die Abfrage wieder in ihren normalen Rhythmus zurückkehren.

Fazit

Jahrelang war die Partitionsanzahl einer zustandsbehafteten Streaming-Abfrage eine Entscheidung, die man einmal ganz am Anfang traf und dann nie wieder anrührte – oder man musste teuer dafür bezahlen, den State von Grund auf neu aufzubauen. Die On-Demand-State-Repartitionierung hebt diese Einschränkungen auf. Die sichere Neuverteilung des States auf eine neue Partitionsanzahl macht eine Entscheidung, die früher nur zum Startzeitpunkt getroffen werden konnte, zu einer Entscheidung, die Sie jederzeit anpassen können, wenn Ihr Workload es erfordert.

Das Ergebnis ist genau das, was sich Betreiber von lang laufenden Streams gewünscht haben: die Freiheit, eine Abfrage basierend auf ihren Skalierungsanforderungen optimal zu dimensionieren – mit nichts weiter als einem Stopp, einer Konfigurationsänderung und einem Neustart, ohne den State zu verlieren.

Die On-Demand-State-Repartitionierung ist in Databricks Runtime 18 und höher unter Verwendung des RocksDB-State-Store-Providers verfügbar. Die vollständige Referenz finden Sie unter On-Demand-State-Repartitionierung für zustandsbehaftete Streaming-Abfragen.

(Dieser Blogbeitrag wurde mit KI-gestützten Tools übersetzt.) Originalbeitrag