Kafka Streams App separat liest aus schreibt

Question

Ich bin ziemlich neu in Kafka und Kafka Streams also bitte mit mir ertragen. Ich würde gerne wissen, ob ich hier auf dem richtigen Weg bin.

Ich schreibe gerade zu einem Kafka-Thema und versuche über einen Rest-Service auf die Daten zuzugreifen. Die Rohdatenart muss transformiert werden, bevor auf sie zugegriffen wird.

Was ich bisher habe, ist ein Produzent, der die Rohdaten in ein Thema schreibt.

1.) Jetzt möchte ich Streams App (sollte ein Glas in einem Container ausgeführt werden), die nur die Daten in meiner gewünschten Form transformiert. Dem materialisierten Ansichtsparadigma folgend.

über vereinfachte Version von 1.)

KStreamBuilder builder = new KStreamBuilder(); 

    KStream<String, String> source = 
    builder.stream("my-raw-data-topic"); 

    KafkaStreams streams = new KafkaStreams(builder,props); 
    KTable<String, Long> t = source.groupByKey().count("My-Table"); 
    streams.start(); 

2.) und einer anderen Strom App (sollte in einem Behälter ein Gefäß laufen), die die KTable als eine Art Repository justs hält, die sein kann Zugriff über einen Wickel-Rest-Service.

Hier bin ich irgendwie fest mit der richtigen Art und Weise mit der API zu arbeiten. Was ist die bloße Minimun auf einen KTable zugreifen und abfragen? Muss ich die Transformationstopologie erneut dem Builder zuweisen?

KStreamBuilder builder = new KStreamBuilder(); 
KTable table = builder.table("My-Table"); //Casting? 
KafkaStreams streams = new KafkaStreams(builder, props); 

RestService service = new RestService(table); 
// Use the Table as Repository which is wrapped by a Rest-Service and gets updated reactivly 

Gerade jetzt ist diese Pseudo-Code

Bin ich auf dem richtigen Weg hier? Ist es sinnvoll zu trennen 1.) und 2.)? Ist dies die eingeprägte Methode, mit Streams zu arbeiten, um Ansichten zu materialisieren? Für mich hätte es den Vorteil, die Schreibvorgänge und die Lesevorgänge über Container unabhängig zu skalieren, wo ich mehr Traffic sehe.

Wie die repopulating der K-Tabelle ist behandelt auf einem Absturz entweder 1.) oder 2.). Ist dies über die Replikation auf die Streaming-API geschehen, oder müsste ich etwas per Code ansprechen? Wie den Cursor zurücksetzen und die Ereignisse beantworten?

Answer 1

Paar Kommentare:

In Ihrem Code-Schnipsel (1) Sie Ihre Topologie ändern, nachdem Sie den Generator in die KafkaStreams Konstruktor übergeben:

KafkaStreams streams = new KafkaStreams(builder,props); 
// don't modify builder anymore! 

Sie sollten dies nicht tun, aber zuerst Sie Topologie angeben und erstellen Sie anschließend die KafkaStreams Instanz.

Über teilen Sie Ihre Anwendung in zwei. Dies kann sinnvoll sein, um beide Teile unabhängig voneinander zu skalieren. Aber es ist im Allgemeinen schwer zu sagen. Wenn Sie jedoch beide spucken, muss der erste das transformierte Datum in ein Ausgabethema schreiben und das zweite sollte dieses Ausgabethema als Tabelle lesen (builder.table("output-topic-of-transformation"), um die REST-Anforderungen zu bedienen.

Für den Laden der K-Tabelle zugreifen, müssen Sie über die mitgelieferten Speichernamen eine Abfrage Griff bekommen:

ReadOnlyKeyValueStore keyValueStore = 
streams.store("My-Table", QueryableStoreTypes.keyValueStore()); 

Lesen Sie die Dokumentation für weitere Details:

http://docs.confluent.io/current/streams/developer-guide.html#interactive-queries