Garantierte Reihenfolge in Apache Kafka

Event Hubs wie Apache Kafka geben eine Ordering Guarantee und versprechen damit, dass die Events in derselben Reihenfolge gelesen werden, wie sie in die Queue geschrieben wurde. Dieser Blog-Post beleuchtet die Details und zeigt die Einschränkungen der garantierten Reihenfolge in Event Hubs im Real-Time Big Data.

1 Reihenfolge in Topics

Sei es als Message Queue, sei es als Event Hub Apache Kafka ist sehr beliebt. Apache Kafka kommt in Realtime Big Data Stream Processing Systemen zum Einsatz. Als verteiltes System ist Kafka jedoch sehr komplex. Das folgende Bild verdeutlicht die Grundidee:

Der Producer schreibt laufend Nachrichten auf eine Topic, also eine Message Queue. Der Consumer holt diese bei Bedarf dort ab.

Zusammen mit der Aussage der Ordering Guarantee, also der garantierten Reihenfolge, suggeriert das Bild, dass der Consumer die Nachrichten in genau der Reihenfolge erhält, wie der Producer sie schreibt. Eine Queue, also Wartschlange, befolgt ja das FIFO-Prinzip – first in first out.

Das folgende Video geht mit einem simplen Test dieser Behauptung nach. Bald stellen wir fest, dass sie einen Trugschluss birgt.

Im ersten Durchlauf des Versuchs sendet der Producer die Events im Abstand von einer Sekunde an das Topic. Der Consumer erhält – wie erwartet – die Events in derselben Reihenfolge, in der der Producer sie versendet hat.

Im zweiten Durchlauf sendet der Producer die Events ohne Wartezeit und der Consumer erhält sie in einer abweichenden Reihenfolge.

Die Erklärung für das Phänomen ist einfach und basiert auf der Tatsache, dass das Topic eine verteilte Message Queue darstellt. Es wurde ja zu Beginn des Experiments mit fünf Partitionen definiert. Der Befehl im Video lautet ja:

kafka-topics.sh --create --topic nugget02 --partitions 5  --replication-factor 2 --bootstrap-server kafka-1:9092

Für das Experiment ist die Angabe –partitions 5 ausschlaggebend. Der Effekt des Experiments ist vergleichbar, wenn das Topic mit 2 oder mehr Partitionen definiert wird.

2 Topics und Partitionen

Als Big-Data-System ist Apache Kafka ein verteiltes System. Die Daten der Topics werden also auf mehrere Broker verteilt, so dass insgesamt sehr große Datenmengen gehandhabt werden können. Je mehr Partitionen ein Topic umfasst, umso mehr Events kann das Topic enthalten.

Bei den Überlegungen zur Partitionierung ist zu berücksichtigen, dass eine Partition nicht über mehrere Festplatten hinweg gehalten werden kann. Der Festplattenplatz beschränkt also die Größe einer einzelnen Partition. Apache Pulsar ist ein Event-Hub, der diese Beschränkung nicht aufweist.

Bei der Entscheidung, wie viele Partitionen für ein Topic definiert werden sollen, spielt auch die Anzahl der vorhandenen Kafka-Broker eine Rolle. Die Partitionen sollten ja auf unterschiedliche Broker verteilt werden, um mehr Events halten zu können. Im Code oben wurden weder retention.ms noch retention.bytes definiert. Die Retention Time bestimmt die Dauer, während der ein Event in der Message Queue verbleibt, bevor es von Kafka automatisch gelöscht wird. Analog bestimmt die retention.bytes wie gross eine Partition maximal werden kann, bevor Kafka, die ältesten Events löscht, um diese Maximalgröße nicht zu überschreiten.

3 Partitionen und Reihenfolge

Topics in Event Hubs sind ja (unbeschränkte) unbounded Datasets. Das bedeutet nicht, dass sie selbst unendlich groß werden – denn der Platz auf der Harddisk ist ja immer beschränkt und auch die Anzahl der Broker ist beschränkt. ‘Unbeschränkt’ bedeutet in dem Fall, dass die Event Queue theoretisch unendlich lang laufen kann – ältere Events werden ja gelöscht, wie oben beschrieben. Bei passender Konfiguration und Pflege wird eine unbeschränkte Laufzeit erreicht.

Jede Topic Partition ist eine Queue, also eine Warteschlange, die nach dem FIFO-Prinzip funktioniert. Damit ist ja die Reihenfolge gegeben: die Events werden in derselben Reihenfolge aus der Queue gelesen, in der sie in die Queue geschrieben wurden.

In einer verteilten Queue verteilt der Producer die Events auf die einzelnen Partitionen. Ein Event wird in genau einer Partition gespeichert. Innerhalb einer Partition werden also die Events in genau derjenigen Reihenfolge aufbewahrt, in der sie geschrieben wurden.

In unserem Experiment werden die Events also auf fünf Partitionen verteilt. Welches Event in welcher Partition landen wird, ist nicht vorhersehbar.

Doch warum stimmt die Reihenfolge im ersten Durchlauf des Experiments und im zweiten Durchlauf nicht mehr?

Der Unterschied war die Geschwindigkeit, in der der Producer die Events auf die Topic-Partitionen schickte. Um den Effekt zu verstehen, schauen wir den Partitioner genauer an.

4 Der Partitioner

Das Kafka-Partitioner API ist einfach bedienbar. Der Python Code im Experiment sah wie folgt aus:

producer = KafkaProducer(bootstrap_servers='kafka-1')
....
producer.send(topic="nugget02", value=data)

Damit wurde das Default-Verhalten von Kafka aufgerufen. Dieses setzt Vieles voraus, was vielleicht gar nicht gewünscht ist. Auch die Wahl des Partitioners.

Es stehen mehrere Partitioner-Algorithmen zur Auswahl.

Der Partitioner definiert einen Buffer für jede Partition, auf die er schreiben soll. Die eintreffenden Events verteilt er nach dem gewählten Algorithmus auf die Partitionen. Beim Round-Robin-Partitioner beispielsweise, werden die Events der Reihe nach auf die einzelnen Buffer verteilt.

Ist ein Buffer voll oder ist eine gewisse Zeit verstrichen, dann werden die Events an die jeweilige Partition gesendet. Beide Parameter können konfiguriert werden.

Dieser Timeout ist natürlich viel kürzer als die eine Sekunde, die wir im ersten Durchlauf des Experiments definiert hatten. Somit wird jedes Event einzeln an die jeweilige Partition gesandt. Im zweiten Durchlauf des Experimants wurden dann mehrere Events gleichzeitig an die Partition gesandt.

Das ist nur ein Teil der Erklärung für die abweichende Reihenfolge, in der der Consumer in den beiden Durchläufen die Events geliefert bekam. Der zweite Teil der Erklärung ist beim Consumer zu suchen.

Dieser pollt die einzelnen Partitionen in einer bestimmten (ebenfalls konfigurierbaren) Frequenz und erhält jeweils höchstens so viele Events geliefert, wie im vorgesehenen Buffer Platz finden (auch diese Parameter in konfigurierbar).

Im ersten Durchlauf des Experiments trafen die Events so langsam in den Partitionen ein, dass sie einzeln ausgeliefert wurden und so die scheinbar richtige Reihenfolge einhalten konnten. Im zweiten Durchlauf des Experiments war das nicht mehr der Fall und die globale Reihenfolge der Events geriet beim Consumer durcheinander.

5 Partition beeinflussen

Bei all diesen Betrachtungen dürfen wir eins nicht außer Acht lassen: Auch beim Consumer wird es sich meistens um ein verteiltes System handeln, beispielsweise um Apache Spark, oder ein anderes Tool zur Real-Time Analyse von Kafka-Topics.

Sollte für diese Verarbeitung oder Analyse der Daten die Reihenfolge eine Rolle spielen, dann wird sich die Analyse-Komponente in ihrer Rolle als Consumer darum kümmern. So wird Apache Spark für Real-Time Analysen die Events in Zeitfenster einteilen und auswerten (siehe Zeit im Big Data Stream Processing).

Verlangt unser Use Case, dass gewisse Events in der richtigen Reihenfolge aus der Queue gelesen werden, dann haben wir bei Apache Kafka die folgenden Möglichkeiten:

5.1 Beim Senden der Events die Partition angeben

Diese Variante sollte mit Bedacht angewendet werden. Immerhin gibt es eine ganze Reihe verschiedener Partitioner-Algorithmen, die genau für diesen Zweck geschrieben wurden.

5.2 Mit nur einer Partition arbeiten

Damit sind wir auf der sicheren Seite: Wo nichts aufgeteilt wird, gerät nichts durcheinander. Doch wir verlieren die Möglichkeit, die Daten auf mehrere Broker zu verteilen und damit die Menge insgesamt zu skalieren.

5.3 Mit Keys arbeiten

Im Code-Beispiel oben, enthielt der Send-Befehl nur ein Value um die Variable zu benennen, in der die zu sendenden Daten hinterlegt sind. Der Send-Befehl kann auch einen Key enthalten. Dabei handelt es sich nicht um einen Primary-Key, sondern lediglich um ein Merkmal, das den Value indentifiziert.

Der Primary-Key könnte beispielsweise eine Konto-Nummer sein. Der zugehörende Value könnte die Veränderungen auf dem Konto enthalten. Für jede Veränderung auf dem Konto wird ein Event ins Topic geschrieben. So können wir die Veränderungen nachverfolgen – immerhin haben sie alle denselben Key, werden in dieselbe Partition geschrieben und erscheinen demnach in der historischen Reihenfolge.

Der Partitioner serialisiert zuerst den Key und bildet danach einen Hash-Wert bilden, dessen Ergebnis die Partition bezeichnet. Damit erwirken wir, dass alle Events mit demselben Key auf ein und dieselbe Partition geschrieben werden und das unter Einhaltung der Reihenfolge ihres Eintreffens. Eine Partition kann für mehrere Keys zuständig sein.

Damit erreichen wir eine sinnvolle Reihenfolge in Bezug auf ein Kriterium: Beispielsweise in Bezug auf die Messdaten eines einzelnen Sensors, wenn die SensorID als Kafka-Key dient.

Die Gefahr besteht, dass die Partitionen ungleichmäßig ausgelastet werden und damit die Consumer der nachfolgenden Komponente (z.B. Apache Spark) ungleichmäßig ausgelastet werden.

Mit Keys arbeiten ist auch sinnvoll, wenn zusätzlich die Topic Partitionen compacted werden. Bei der Compaction bleibt jeweils das jüngste Element eines Keys erhalten und die älteren werden gelöscht. Eine sinnvolle Art Daten mit relativ wenigen unterschiedlichen Keys in Topics zu verwalten.

6 Fazit

Apache Kafka gibt eine Ordering Guarantee ab. Gleich wie andere verteilte Event-Hubs kann diese Garantie nur in Bezug auf eine Partition abgegeben werden.

Durch die Wahl eines Keys kann die Reihenfolge der Events im Topic in Bezug auf den Key garantiert werden.