Apache Flink: Vollständiges Checkpoint vs. Inkrementelles Checkpoint

Apache Flink ist eine Echtzeit-Datenstromverarbeitungs-Engine. Die meisten Stream-Verarbeitungsanwendungen sind „zustandsbehaftet“. Das bedeutet, dass der Zustand gespeichert und für die weitere Verarbeitung verwendet wird. In Apache Flink wird der Zustand über ein konfiguriertes State Backend verwaltet. Flink unterstützt in der Produktion zwei State Backends. Eines ist das HashMapStateBackend, und das andere ist das EmbeddedRocksDBStateBackend. 

Um Datenverlust zu verhindern und Fehlertoleranz zu erreichen, kann Flink Snapshots des Zustands in einem dauerhaften Speicher persistieren. Flink kann so konfiguriert werden, dass entweder der gesamte Zustand an einem dauerhaften Ort oder das Delta seit dem letzten Snapshot gespeichert wird. Letzteres wird als vollständiger Checkpoint bezeichnet, während ersteres als inkrementeller Checkpoint bekannt ist. 

In diesem Artikel werden wir HashMapStateBackend mit vollem Checkpoint und EmbeddedRocksDBStateBackend mit inkrementellem Checkpointing vergleichen. Dieser Artikel geht davon aus, dass das Publikum entweder praktische Kenntnisse oder theoretische Kenntnisse über Apache Flink hat. 

Überblick über das Flink State Backend

Um das Flink State Backend zu verstehen, ist es wichtig, den Unterschied zwischen In-Flight-Zustand und Zustands-Snapshots zu kennen. 

Der In-Flight-Zustand ist als Flinks Arbeitszustand bekannt und wird lokal gespeichert. Je nach Konfiguration des State Backends befindet er sich entweder im Heap-Speicher oder im Off-Heap-Speicher, mit einer möglichen Auslagerung auf die lokale Festplatte.

Auf der anderen Seite werden Zustandsschnappschüsse (Checkpoint oder Speicherpunkt) an einem dauerhaften entfernten Ort gespeichert. Diese Schnappschüsse werden verwendet, um den Flink-Jobzustand im Falle eines Jobfehlers wiederherzustellen. 

Der im Flug befindliche Zustand kann verloren gehen, wenn der Job fehlschlägt. Dies hat keinen Einfluss auf die Jobwiederherstellung, wenn der Checkpoint im Job aktiviert ist. Wenn der Checkpoint konfiguriert ist, wird der Zustand zum Zeitpunkt der Wiederherstellung aus dem dauerhaften Speicher abgerufen. 

Welcher Zustandsspeicher für die Produktion ausgewählt werden soll, hängt von den Anforderungen der Anwendung an Durchsatz, Latenzzeit und Skalierbarkeit ab.

Es gibt zwei Zustandsspeicher, die von Apache Flink in der Produktion unterstützt werden.

1. HashMapStateBackend

Es handelt sich um einen leichten Zustandsspeicher in Flink zur Verwaltung des Keyed-State und Operator-State während der Streamverarbeitung. Der Zustand wird im Java-Heap mithilfe einer HashMap-Datenstruktur gespeichert. Da er im Speicher gespeichert ist, ist die Hauptbeschränkung hier, dass die maximale Zustandsgröße auf die Größe des Java-Heaps begrenzt ist. Es gibt keine Serialisierung beim Schreiben in den Zustand oder Lesen aus dem Zustand. Daher eignet sich dies für Anwendungen mit geringer Latenz, hoher Durchsatzrate und nicht allzu großem Zustand. 

2. EmbeddedRocksDBStateBackend

Dieser Zustands-Backend speichert die im Flug befindlichen Daten in der im Arbeitsspeicher befindlichen RocksDB-Datenbank. Standardmäßig speichert RocksDB die Daten auf der lokalen Festplatte des Task-Managers. Die Daten werden serialisiert und in einem Off-Heap-Speicher gespeichert und auf eine lokale Festplatte übertragen, die mit dem Task-Manager verbunden ist. Das Serialisierungsformat hängt vom im Anwendungscode konfigurierten Typ-Serializer ab. 

Mit diesem Zustands-Backend ist die Menge an speicherbarem Zustand nur durch den Festplattenspeicher begrenzt, der dem Task-Manager zugeordnet ist. Wenn die Anwendung einen großen Zustand hat, der nicht im Heap-Speicher enthalten werden kann, ist dies das richtige Zustands-Backend. Da Serialisierung involviert ist, wird die Anwendung im Vergleich zum HashMapStateBackend eine höhere Latenz und eine niedrigere Durchsatzrate haben.

Überblick über den Snapshot-Zustand

Der Snapshot repräsentiert den globalen Zustand des Flink-Jobs. Dies umfasst einen Zeiger auf jeden Datenquellen und den Zustand aller Flink-Zustandsoperatoren nach der Verarbeitung bis zu diesen Zeigern von den Quellen. Das Checkpointing in Apache Flink ist ein Mechanismus zur Erreichung der Ausfallsicherheit durch periodisches Speichern des Zustands in dauerhaftem Remote-Speicher. 

Im Falle eines Jobfehlers ruft Flink den gespeicherten Zustand aus dem dauerhaften Remote-Speicher ab und beginnt mit der Verarbeitung der Streaming-Daten, wo es aufgehört hat. Flink verwendet asynchrones Barrieren-Snapshotting. Es handelt sich um eine Variante des Chandy-Lamport-Algorithmus. 

Flink unterstützt zwei Arten des Checkpointings.

1. Vollständiges Checkpointing

Vollständiges Checkpointing ist, wenn der gesamte Zustand des Flink-Jobs erfasst und in speicherhaltigem Remote-Speicher gespeichert wird. Im Falle eines Jobfehlers wird der Job aus dem zuvor gespeicherten Zustand wiederhergestellt. Die benötigte Speicherkapazität und die für das Checkpointing benötigte Zeit hängen vollständig vom Anwendungszustand ab. Das vollständige Checkpointing funktioniert sowohl mit HashMapStateBackend als auch mit RocksDBStateBackend.

2. Inkrementelles Checkpointing

Inkrementelles Checkpointing ist ein optimierter Ansatz. Anstatt den gesamten Zustand zu snapshotten, speichert Flink nur die seit dem letzten Checkpoint vorgenommenen „Deltas“. Dies reduziert die Netzwerkbelastung und damit die für das Checkpointing benötigte Zeit. In diesem Fall erfolgt das Checkpointing vollständig asynchron.

Nur RocksDBStateBackend unterstützt das inkrementelle Checkpointing. Flink nutzt den internen Mechanismus von RocksDB dafür. Obwohl das Checkpointing weniger Zeit in Anspruch nimmt als das vollständige Checkpointing, hängt die Wiederherstellungszeit im Falle eines Jobfehlers von vielen Faktoren ab. Wenn das Netzwerk ein Engpass ist, kann die Wiederherstellungszeit höher sein als die Wiederherstellung aus dem vollständigen Checkpointing.

Analyse

Pipeline-Details: Apache Beam-Pipeline, die auf der Flink-Engine mit einem festen Fenster von 10 Minuten läuft und das Checkpointing so konfiguriert ist, dass es alle 3 Minuten ausgeführt wird. Der konfigurierte Serialisierungstyp ist AVRO.

• Cluster-Typ: „m5dn.4xlarge“
• Endgültiger Checkpoint-Speicher: S3
• Anzahl der eindeutigen Schlüssel: 2K

Wie aus dem obigen Experiment ersichtlich ist, verringert sich die Checkpoint-Dauer mit inkrementellem Checkpointing. Dies kann die Anwendungsleistung durchaus verbessern.  

Zusammenfassung

Hier ist die Zusammenfassung des Experiments.