TensorFlow mit Java

"Geht nicht!", wird der geneigte Leser in anbetracht der Überschrift nun vielleicht denken. Und auch wenn es nicht so komfortabel ist wie mit Python, es geht doch: TensorFlow von Java aus aufrufen.

Wenn Sie zu den Lesern gehören sollten, die nun fragen, warum man so etwas tun sollte, können Sie sich das Weiterlesen vermutlich sparen. Ich finde es sehr praktisch, dass man den Einstieg in Googles Machine Learning auch als Java-Programmierer finden kann, ohne sich zusätzlich in eine neue Programmiersprache einarbeiten zu müssen. Wer hier allerdings tiefer einsteigen und bspw. eigene Modelle trainieren möchte, wir über kurz oder lang um Python vermutlich nicht herum kommen, da die Unterstützung für Java leider eher dürftig ist.

Das Einbinden von TensorFlow in ein Java-Progamm ist dank folgender Maven-Dependency jedenfalls hinreichend einfach:

   <dependency>
      <groupId>org.tensorflow</groupId>
      <artifactId>tensorflow</artifactId>
      <version>1.13.1</version>
  </dependency>

Alternativ lässt sich natürlich auch das entsprechende JAR-File herunterladen und direkt in den Classpath legen. Die aktuelleste Version ist übrigens 1.14.0, diese läuft auf meinem Mac allerdings nicht, so dass ich auf 1.13.1 zurückgegriffen habe, was dann auch problemlos funktioniert.

Das Kopieren des reichlich nichtssagenden Einstiegsbeispiels von der TensorFlow-Webseite spare ich mir an dieser Stelle, wer damit testen möchte, kann es leicht von dort kopieren. Bevor ich gleich ein etwas nachvollziehbareres Beispiel zum Besten gebe, möchte ich an dieser Stelle aber doch ein Worte zu den Hintergründen verlieren.

Wie die allermeisten Leser sicher bereits gehört haben, ist TensorFlow ein von Google entwickeltes und als Open-Source verbreitetes Framework für Machine Learning, das sich auf Grund seiner vielseitigen Einsetzbarkeit (sowohl was die Einsatzmöglichkeiten, als auch was die Plattformen angeht) einer großen Beliebtheit erfreut.

Von der Grundidee her, verwendet TensorFlow (eine Art Datenfluss-)Graphen, um Daten (also die Tensoren) zu verarbeiten. Ein Tensor ist übrigens nichts weiter als ein Array von Daten mit einer quasi beliebigen Anzahl von Dimensionen. Im einfachsten Fall mit null Dimensionen, womit es sich  um nichts weiter als einen Skalar (also eine Zahl) handelt. Ein eindimensionales Array ist eine Liste von Daten, ein zweidimensionales eine Tabelle usw...

Wir können uns die Graphen in Tensorflow ein wenig wie Aktivitätsdiagramme vorstellen oder auch wie die in modernen Streaming-Frameworks verwendeten Graphen zur Beschreibung der gewünschten Datenverarbeitung. Genau zu diesem Zweck werden Sie auch in TensorFlow verwendet. Entsprechend muss in einem Java-Programm auch zunächst ein Graph als Objekt angelegt werden:

  Graph g = new Graph();

Diesen Graphen können wir nun Schritt für Schritt zusammenbauen, wie das geht, wollen wir uns zunächst am Beispiel einer einfachen Addition betrachten. Dazu legen wir mit Hilfe der OperationBuilder-Methode des Graphen Operanden für zwei Skalare an. Diese werden durch den Aufruf der zugehörigen Build-Methode im Graphen verfügbar gemacht:

  Operation x = g.opBuilder("Placeholder", "x").setAttr("dtype", DataType.DOUBLE).build();     
    
  Operation y = g.opBuilder("Placeholder", "y").setAttr("dtype", DataType.DOUBLE).build();

Um diese beiden einzelnen Knoten zu einem echten Graphen zu verbinden, benötigen wir noch unsere Additionsoperation, die die beiden Placeholder (bzw. die darin enthaltenen Daten) zusammenführt:

  Operation z = g.opBuilder("Add", "sum").addInput(x.output(0)).addInput(y.output(0)).build();

Bis hierher ist noch keinerlei Berechnung passiert, wir haben nur eine leere Hülse für den Graphen angelegt. Um diese mit Leben zu füllen und die eigentliche Berechnung auszuführen, benötigen wir noch eine Session, die wieder ganz normal als Java-Objekt instanziiert wird und dabei den auszuführenden Graphen als Parameter übergeben bekommt:

  Session s = new Session(g); 

Im Anschluss definieren wir als Rückgabe wie folgt einen parametrisierten Tensorund lassen seinen Double-Wert auf der Konsole ausgeben:

  Tensor<Double> t = s.runner().fetch("sum")
      .feed("x", Tensor.<Double>create(4.0, Double.class))
      .feed("y", Tensor.<Double>create(6.0, Double.class))
      .run().get(0).expect(Double.class);

  System.out.println(t.doubleValue());

  t.close();
  s.close(); 

Wie unschwer am Beispiel-Code zu erkennen ist, wird sum als Ergebnis nach Ausführen der Additionsoperation abgegriffen, vor der Ausführung müssen natürlich noch Werte für die beiden Platzhalter x und y an den Session-Runner gefüttert werden. Abschließend sollten sowohl der angelegte Tensor als auch die Session wieder geschlossen werden.

Matrix mit Skalar multiplizieren

Um auch noch ein anspruchsvolleres Beispiel zu demonstrieren, möchte ich im Folgenden erklären, wie sich mit TensorFlow eine Matrix mit einem Skalar multiplizieren lässt. Dazu legen wir zunächst  eine einfache Matrix als Tensor an und legen diesen als Konstante im Graphen ab:

  Tensor<Integer> t2 = Tensor.create(new int[][]{ {5,2},{3,4} }, Integer.class);            
  Operation a = g.opBuilder("Const", "u").setAttr("dtype", DataType.INT32).setAttr("value", t2).build();


Entsprechend können wir mit einem Skalar verfahren:
         
  Operation b = g.opBuilder("Const", "v").setAttr("dtype", DataType.INT32).setAttr("value", Tensor.<Integer>create(4, Integer.class)).build();

Abschließend müssen wir noch die beiden Knoten noch mit einer Multiplikation verbinden, diese über den Session-Runner ausführen lassen und den Ergebnistensor, der natürlich wieder eine Matrix ist, in ein Java-Array umwandeln lassen:           

  g.opBuilder("Mul", "prod").addInput(b.output(0)).addInput(a.output(0)).build();
           
  Tensor<Integer> t3 = s.runner().fetch("prod").run().get(0).expect(Integer.class);
            
  System.out.println(t3.toString());
  int[][] res = t3.copyTo(new int[2][2]);
  System.out.println(res[0][0] + " " + res[0][1]);
  System.out.println(res[1][0] + " " + res[1][1]);

Das war "quick and clean" eine erste Einführung in TensorFlow für/mit Java, weitere Beispiele werden bei nächster Gelegenheit sicherlich folgen.

Beispielcode

Der Vollständigkeit halber sei abschließend noch das vollständige Additionsprogramm mit allen zugehörigen Imports gepostet:

import org.tensorflow.DataType;
import org.tensorflow.Graph;
import org.tensorflow.Operation;
import org.tensorflow.Session;
import org.tensorflow.Tensor;

public class TfAdd {

    public static void main(String[] args) {

        try {
            Graph g = new Graph();

            Operation x = g.opBuilder("Placeholder", "x").setAttr("dtype", DataType.DOUBLE).build();       
            Operation y = g.opBuilder("Placeholder", "y").setAttr("dtype", DataType.DOUBLE).build();
          
            Operation z = g.opBuilder("Add", "sum").addInput(x.output(0)).addInput(y.output(0)).build();

            Session s = new Session(g);

            Tensor<Double> t = s.runner().fetch("sum")
                    .feed("x", Tensor.<Double>create(4.0, Double.class))
                    .feed("y", Tensor.<Double>create(6.0, Double.class))
                    .run().get(0).expect(Double.class);

            System.out.println(t.doubleValue());
            
            t.close();
            s.close();
          
        } catch(Exception e) {
            e.printStackTrace();
        }
    }

} 

Auch die Matrixmultiplikation lässt sich ohne Probleme in diesen Rahmen einbauen. Und als kleine Übungsaufgaben bieten sich natürlich Matrix-Vektor- oder auch Matrix-Matrix-Multiplikationen an.

Stream-Verarbeitung und Windows in Kafkas KSQL

Vor einigen Tagen hatte ich über die Grundlagen von KSQL als Schnittstelle zur Stream-Verarbeitung mit Kafka geschrieben. Die Beispiele dort sind zum Einstieg sehr stark an eine "klassische" relationale Sicht angelehnt. Eigentliches Ziel von und Hauptargument für KSQL ist aber sicherlich die Verarbeitung von "endlosen" Datenströmen über eine erweiterte SQL-Syntax. Und weil diese Möglichkeiten so schön und spannend sind, möchte ich darauf in diesem Post näher eingehen.

Hintergründe

Ein erster einfacher Use Case für die Stream-Verarbeitung mit KSQL könnte bspw. das Filtern von Daten in einem Kafka-Topic sein. Betrachten wir uns dazu noch einmal das Beispiel mit den Studierenden und ihren Prüfungsergebnissen aus meinem einleitenden Blog-Post zum Thema.

Einträge von Studierenden bestehen aus Vorname, Name, Matrikelnummer und Studiengang(skürzel) und können direkt als JSON über den sog. Konsole-Producer (Details zum Aufruf - na wo wohl - im Vorgänger-Post) in Kafka eingespielt werden:

55555:{"vorname":"Steven","name":"Müller-Mayer","matrikelnummer":"55555","studiengang":"IB"}

Zur Erinnerung, Daten werden als Key-Value-Paare in Kafka-Topics gestellt, der Topic-Key sollte zur Weiterverarbeitung als Stream oder Tabelle dem Primärschlüssel entsprechen.

Mit Prüfungsdaten funktioniert das analog, die Matrikelnummer ist hier der Fremdschlüssel, um den entsprechenden Studierenden auflösen zu können:

1:{"id":"1","name":"Big Data","note":1.0,"matrikelnummer":"12345"}

Wie im vorherigen Post gezeigt, legen wir für die Stammdaten der Studierenden ganz klassisch eine Tabelle an. Die (für dieses Beispiel) quasi fortlaufend hereinkommenden Prüfungsergebnisse definieren wir in KSQL als einen Stream von Events.  

Einfache Stream-Verarbeitung

Betrachten wir uns zum Einstieg in die Verarbeitung von Streams zunächst nur eine Select-Anfrage, die die beste Note für eine bestimmte Prüfung ermittelt, das könnte beispielsweise hilfreich sein, um eine Notenübersicht mit Anzahl der jeweils benötigten Versuche für das Zeugnis zu generieren (ich habe dafür noch zwei weitere Prüfungsergebnisse in Kafka eingespielt):

select id, min(note), name, matrikelnummer, count(note) from str_exams where matrikelnummer = '12345' group by id, matrikelnummer, name;
-->
1 | 1.0 | BDEA | 12345 | 1
37 | 4.0 | Programmierung 2 | 12345 | 2

Eine weitere einfache Anwendung auf diesen Daten könnte nun beispielsweise sein, sehr gute Studierende bzw. zunächst Ihre Prüfungsergebnisse herauszufiltern, um ihnen etwa Tutoren-Verträge im jeweiligen Fach anbieten zu können. Dazu leiten wir aus unserem Stream einfach folgendermaßen einen neuen Stream ab:

select name, note, matrikelnummer from str_exams where note < 1.5;

Möchten wir auf das gefilterte Ergebnis wieder als eigenständigen Stream zugreifen können ist das sehr einfach durch folgende fettgedruckte Ergänzung möglich:

create stream great_exams as select name, note, matrikelnummer from str_exams where note < 1.5;

Ein einfaches Select auf diesem Stream liefert uns daraufhin das bisher einzige sehr gute Prüfungsergebnis:

select * from great_exams;
--> 1535386252767 | 1 | BDEA | 1.0 | 12345

Um auch die Namen der Studierenden ausgeben zu können, ist offensichtlich ein Join mit der Tabelle der Studierendendaten erforderlich:

create stream great_exams as select str_exams.name, note, str_exams.matrikelnummer, studis.name from str_exams left join studis on str_exams.matrikelnummer = studis.matrikelnummer where note < 1.5;

Löschen von Streams

Da der zuerst angelegte Stream wieder gelöscht werden muss, bevor die Variante mit dem Join unter gleichem Namen angelegt werden kann, an dieser Stelle eine kurze Zwischenbemerkung zum Löschen von Streams. Es funktioniert wie gewohnt per drop stream great_exams; allerdings muss zuvor noch die laufende Streaming-Query wie folgt terminiert werden:

terminate CSAS_GREAT_EXAMS;

Vergisst man das Terminieren vor dem Drop beschwert sich KSQL entsprechend, liefert praktischerweise den Namen der Query in der Fehlermeldung direkt mit, so dass dieser einfach kopiert werden kann.

Ein weitere Besonderheit zeigt sich allerdings noch nach einer Select-Abfrage auf dem gejointen Stream, hier taucht nach dem neu Anlegen sonderbarerweise ein Ergebnis mit null-Einträgen auf:

select * from great_exams;
-->
1535386252767 | 1 | null | 1.0 | null | null
1535386252767 | 12345 | BDEA | 1.0 | 12345 | Mayer

Das liegt schlicht daran, dass KSQL für jeden neu angelegten Stream ein Topic zur Datenspeicherung benötigt, dieses Topic automatisch nach dem Stream-Namen benennt und es beim Löschen des Streams nicht mitlöscht. Folglich bleibt in diesem Fall das "ungejointe" Ergebnis aus der ersten Select-Abfrage im Topic liegen und führt nun zu den Null-Einträgen.

Das Löschen von Topics funktioniert in der aktuellen Version 5.0.0 der Confluent-Plattform noch nicht über KSQL, das geht bis dato nur mit dem aktuellen Preview-Release durch Ergänzen von delete topic nach dem Drop (vgl. hier) oder direkt über Kafka. Alternativ lässt sich der Name des Topics über den Parameter kafka_topic beim Erstellen eines Streams bzw. einer Tabelle auch selbst festlegen.

 

Weitere Abfragen

Mit einem Join ist nun bspw. auch eine Filterung der Prüfungsergebnisse nach Studiengang leicht machbar:

create stream imb_exams as select * from str_exams left join studis on str_exams.matrikelnummer = studis.matrikelnummer where studis.studiengang = 'IMB';

Bisher haben wir aus existierenden Streams abermals Streams abgeleitet, da wir den Input-Stream nur gefiltert haben. Aggregieren wir Daten auf einem Stream, wird die Ergebnissemantik eine andere. Dann interessieren uns nicht mehr einzelne Events, sondern wir sind an der Zusammenfassung mehrerer Ereignisse interessiert. Diese werden in KSQL in eine Tabelle persistiert.

Wie das folgende Beispiel verdeutlicht, ist auch das nicht weiter schwierig: angenommen wir möchten alle Studierenden, die eine Prüfung zum dritten Mal nicht bestanden haben, als endgültig durchgefallen markieren, können wir sie folgendermaßen in einer Tabelle speichern (und auf dieser wieder beliebige SQL-Abfragen ausführen):

create table finally_failed as select matrikelnummer, name, count(*) from str_exams  group by matrikelnummer, name having min(note) = 5.0 and count(*) > 2;

Windowing

Das war aber noch längst nicht alles, was KSQL zu bieten hat: wer sich bereits mit der Verarbeitung von Daten-Streams beschäftigt hat, weiß, dass es meist spannend ist, Events innerhalb von Zeitfenstern zu analysieren. Beispielsweise wird sicherlich jedes gut genutzte Kreditkarten-Konto über seine Lebensdauer viele Transaktionen erfahren, aber viele Transaktionen innerhalb einer Zeitspanne von bspw. 10 min, könnten auf Betrugsversuche hindeuten. Um entsprechende Abfragen komfortabel ausführen zu können, muss das Stream-Verarbeitungs-Framework die Definition von Windows unterstützen. Und natürlich bietet auch KSQL diese Möglichkeit (vgl. z.B. dieses Video ab Minute 1:50).

Betrachten wir als Beispiel die im Hochschul-Alltag zugegebenermaßen etwas konstruierte Fragestellung, ob es Dozenten/Bearbeiter gibt, die Noten einer Prüfung innerhalb eines kurzen Zeitrahmens häufig ändern. In diesem Fall möchten wir in einem neuen Stream darüber informiert werden. Die eigentliche SQL-Abfrage dazu gestaltet sich relativ einfach:

select id, name, matrikelnummer from str_exams group by id, name, matrikelnummer having count(*) > 2;

Um die entsprechende Eingrenzung auf einen Zeitraum von bspw. einer Minute zu erreichen, definieren wir nun noch ein entsprechendes Zeitfenster:

select id, name, matrikelnummer from str_exams window tumbling (size 1 minute) group by id, name, matrikelnummer having count(*) > 2;

Auch eine Tabelle lässt sich davon wieder problemlos ableiten:

create table changes as select id, name, matrikelnummer from str_exams window tumbling (size 1 minute) group by id, name, matrikelnummer having count(*) > 2;

KSQL bietet drei Arten von Fenstern an, das bereits gezeigte Tumbling Window, das nicht überlappende Abschnitte des definierten Zeitraums umfasst, das sog. Hopping Window, das die Fenster immer eine bestimmte Zeitspanne überlappen lässt oder sog. Session Windows, die alle Events zusammenfassen, die nicht durch eine Pause bestimmter Länge unterbrochen werden. Für letzteres lässt sich obige Anfrage z.B. wie folgt formulieren:

select id, name, matrikelnummer from str_exams window session (1 minute) group by id, name, matrikelnummer having count(*) > 2;

Noch ein Wort zu Timestamps

Abschließend noch einige Gedanken zu Timestamps in einer verteilten Big-Data-Architektur. Hier kann es natürlich passieren, dass Daten Kafka erst mit einer gewissen Verzögerung erreichen, der Kafka-Timestamp also nicht geeignet für die weitere Analyse ist. Per Default verwendet Kafka eine sog. Retention Time von einem Tag innerhalb derer Fenster noch mit später kommenden Daten aktualisiert werden. Dazu ist üblicherweise die Verwendung der sog. Event Time, also eines Zeitstempels, der im Ursprungssystem generiert wird, notwendig. Das funktioniert schlicht über die Angabe des Parameters TIMESTAMP='spaltenname' beim Anlegen eines Streams, wobei der genannte Spaltenname einen Zeitstempel in den Quelldaten referenzieren muss. Den Rest erledigt Kafka automatisch.

Kafka + Streaming + SQL = KSQL

Apache Kafka dürfte vielen Entwicklern vor allem als schnelles Queuing-System und als Quasi-Standard in diesem Bereich bekannt sein. Entsprechend wird es bisher vor allem als Puffer verwendet, um ankommende Daten vor der Verarbeitung zwischenzuspeichern oder einen möglichen Rückstau von Zwischenergebnissen und damit einen Nachrichtenverlust bei der Datenweitergabe zwischen verschiedenen Services zu vermeiden. 

Kafka-Mastermind Jay Kreps hat allerdings große Pläne für sein Werkzeug, mit dem Einbau von Stream-Verabeitungsmöglichkeiten und sogar einer KSQL genannten SQL-Schnittstelle, plant er mit Kafka nichts weniger, als Datenbanken neu zu erfinden (vgl. auch: Turning the Database inside out) und hat damit bereits einiges an Neugierde ausgelöst. In diesem Blog-Beitrag möchte ich daher einige erste Ideen und Möglichkeiten aber auch aktuelle Beschränkungen von KSQL diskutieren und wie alle Schritte nachvollziehbar auflisten, die zum direkten Ausprobieren notwendig sind.

 

Getting Started

Der einfachste Weg KSQL auszuprobieren, ist sicher die Verwendung der sog. Confluent Platform (Download hier), die alle dafür benötigten Einzelteile intergriert. Confluent (engl. für zusammenfließend bzw. -wachsend) ist übrigens die Firma, die Jay Kreps mit Kollegen nach seinem Weggang bei LinkedIn gegründet hat, um Kafka zu vermarkten.

Voraussetzung zum Betrieb der Plattform ist ein installiertes Java 8 (neuere Versionen funktionieren aktuell leider noch nicht), dann kann sie nach dem Herunterladen und Entpacken direkt auf der Kommandozeile aus dem bin-Ordner durch Aufrufen von ./confluent start ausgeführt werden. Danach steht die KSQL-Shell per ./ksql zur Verfügung. SQL-Entwickler sollten sich hier schnell heimisch fühlen, da die üblichen SQL-Befehle wie gewohnt funktionieren, bspw. show tables;

Um eine Tabelle anlegen zu können, werden in KSQL allerdings sog. Kafka-Topics benötigt, in die die gewünschten Daten eingespielt werden. Für alle, die Kafka bisher nicht kennen, Topics sind Kafkas Ansatz zur Datenspeicherung und zunächst im Prinzip nichts weiter als große Listen bestehend aus Key-Value-Paaren. Um Daten in ein solches Topic hineinzubekommen gibt es, wer hätte es gedacht, ein weiteres Kommandozeilen-Skript (kafka-console-producer), das wie folgt aufgerufen werden kann:

./kafka-console-producer --broker-list localhost:9092 --topic exams --property "parse.key=true" --property "key.separator=:" < /pfad/prüfungen.json

Der Einfachheit halber pipen wir die Daten, die wir in das Topic hineinlegen möchten, aus einer Datei names prüfungen.json heraus. Diese könnte bspw. folgende Daten für Prüfungsleistungen an einer Hochschule oder Universität enthalten:

1:{"id":"1","name":"Big Data","note":1.0,"matrikelnummer":"12345"}
5:{"id":"5","name":"Programmierung 1","note":2.0,"matrikelnummer":"23456"}
17:{"id":"17","name":"Programmierung 1","note":4.0,"matrikelnummer":"12345"}

Wie leicht erkennbar ist, enthält die Datei drei Zeilen mit Key-Value-Paaren, die jeweils durch einen Doppelpunkt unterteilt werden. Der Value besteht aus einem JSON-Eintrag, der die eigentlichen Prüfungsdaten enthält. In obigem Beispiel sind das eine ID, ein Vorlesungsname, die erreichte Note sowie die Matrikelnummer des Studierenden, der die Prüfung abgelegt hat, als Fremdschlüssel. Der Key des Key-Value-Paares sollte dabei immer dem Primärschlüssel des Values, in unserem Fall also der Prüfungs-ID, entsprechen.

Nun können wir in der KSQL-Shell nachschauen, ob alles korrekt funktioniert hat, in dem wir uns zunächst alle Topics ausgeben lassen: show topics;

Wir haben damit (endlich) eine schöne Möglichkeit, die in Kafka enthaltenen Daten zu erkunden. Taucht dort nämlich das Topic mit dem zuvor gewählten Namen exam auf, können wir uns auch seinen Inhalt folgendermaßen anzeigen lassen: print 'exams' from beginning;

Doch... die Abfrage scheint hängen zu bleiben..? DON'T PANIC, it's not a bug, it's a feature. Da Topics im Real-Betrieb normalerweise fortlaufend mit Daten gefüttert werden, sind zunächst alle Queries in KSQL sogenannte Continuous Queries, die so lange laufen und neu in das Topic kommende Daten ausgeben, bis sie abgebrochen werden. Das geht wie auf der Kommandozeile üblich mit Control + C.

Nun haben wir alle Vorbereitungen getroffen, um in KSQL wie folgt eine Tabelle anlegen zu können:

create table exams (id varchar, name varchar, note double, matrikelnummer varchar) with (kafka_topic = 'exams', value_format='json', key = 'id');

Der erste Teil ist dabei (fast) normaler SQL-Code, im zweiten Teil werden einige Metadaten übergeben, die KSQL mitteilen, mit welchem Topic die Tabelle verbunden werden soll.

 

(K)SQL-Abfragen

Wie von SQL gewohnt können wir uns nun das Schema über describe exams; anzeigen lassen, dabei wird deutlich, dass Kafka automatisch zwei weitere Spalten zu unseren Daten hinzugefügt hat, nämlich die ROWTIME und den ROWKEY. Erstere ist der Timestamp, wann Kafka den Datensatz empfangen hat, zweitere ist der Key des jeweiligen Key-Value-Paars aus dem Kafka-Topic, der mit dem Inhalt der Key-Spalte, die in der With-Klausel genannt ist identisch sein muss. Ferner können wir uns natürlich auch den Inhalt der Tabelle mit einer normalen Select-Abfrage ausgeben lassen:

select * from exams;

Doch Moment, wir sehen, dass wir außer einer "hängenden Query", die sich mit Ctrl-C beenden lässt, wieder nichts sehen...? Das liegt daran, dass Kafka per default so konfiguriert ist, dass es nur neu ankommenden Einträge in einem Topic auswertet (daher zuvor auch das "from beginning"). Wir könnten also a) neue Einträge an das Topic schicken oder b), wenn wir alle vorhandenen Einträge in der Select-Abfrage angezeigt bekommen möchten, müssen wir vor der nächsten Anfrage folgendes Kommando auf der KSQL-Shell absenden (und das nach jedem Neustart der Shell):

SET 'auto.offset.reset' = 'earliest'; 




Nun sollte unsere Select-Abfrage problemlos funktionieren, daher können wir auch gleich noch eine Where-Klausel an sie anhängen:

select * from exams where name = 'Programmierung 1';

Einer Query kann auch ein Limit mitgegeben werden, so dass sie automatisch endet, wenn die gesetzte Anzahl von Ergebnissen ausgegeben worden ist:  select * from exams where name = 'Programmierung 1' limit 1;

Ferner sind natürlich auch Aggregat-Funktionen wie z.B. die Berechnung des Notendurchschnitts pro Fach möglich (auch wenn es bis dato keine avg-Funktion gibt):

select name, sum(note) / count(name) from exams group by name;

Dieser Aufruf ergibt folgende Ausgabe:

Big Data | 1.0
Programmierung 1 | 3.0

Interessant dabei ist es nun zu beobachten, was auf der nicht abgebrochenen Streaming Query passiert, wenn ein weiteres Prüfungsergebnis in das Kafka-Topic geschrieben wird. Das können wir bspw. tun, in dem wir den Konsole-Producer wie oben nur ohne Piping aufrufen und dann folgende Zeile dort hinein pasten:

19:{"id":"19","name":"Big Data","note":2.0,"matrikelnummer":"55555"}

Dadurch wird das Ergebnis der noch laufenden Query aktualisiert und ein neuer Durchschnitt für das Fach Big Data ausgeben:

Big Data | 1.5

Analog würde der neue Eintrag auch bei einer einfachen Select-Query angehängt werden. Übrigens unterstützt es Kafka auch direkt, den Timestamp von Einträgen für Menschen lesbar zu formatieren: select TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss.SSS') from exams;

 

Tables vs. Streams

Neben Tables unterstützt KSQL auch sogenannte Streams, die ebenfalls per SQL durchsucht werden können. Während Tabellen immer den aktuellen Stand der Datenlage, also quasi einen Snapshot, ausgeben, speichern und zeigen Streams immer alle aufgetretenen Events. Schauen wir uns als Beispiel die Änderung einer Note an, die wir über den Konsole-Producer einspielen können:

19:{"id":"19","name":"Big Data","note":1.7,"matrikelnummer":"55555"}

Eine Select-Abfrage auf der Tabelle liefert für diesen Key genau den zuletzt eingetragenen Stand der Note, bzw. natürlich auch den aller enthaltenen Spalten:

select * from exams where id='19'; 

   --> 1535127212852 | 19 | 19 | Big Data | 1.0 | 55555

Wie sieht das im Gegensatz dazu bei einem Stream aus? Auch diesen müssen wir zunächst einmal anlegen:

create stream str_exams (id varchar, name varchar, note double, matrikelnummer varchar) with (kafka_topic = 'exams', value_format='json', key = 'id');

Danach können wir die entsprechende Abfrage losschicken:

select * from str_exams where id='19';

   --> 1535126428549 | 19 | 19 | Big Data | 2.0 | 55555
       1535127660012 | 19 | 19 | Big Data | 1.7 | 55555

... und erhalten hier die gesamte Stream-Historie unseres Datensatzes. Ein offensichtlicher Usecase im vorliegenden Fall ist daher die Historisierung von Daten, die richtig spannende Möglichkeit liegt hier allerdings darin, dass eine Verarbeitung von Streaming-Daten mit Hilfe von SQL-Befehlen möglich wird, also kein eigenes Streaming Framework wie Flink oder Storm mehr notwendig ist. Darauf werde ich sicher bei nächster Gelegenheit noch genauer eingehen.

Sowohl Tabellen als auch Streams können übrigens wie gewohnt über drop table bzw. stream plus den Namen wieder gelöscht werden.

 

Limitierungen

Beenden wollen wir unseren ersten Ausflug in die Welt von KSQL, allerdings mit der Frage, wie es um die Join-Fähigkeiten der aktuellen KSQL-Version bestellt ist. Grundsätzlich sind Joins zwar vorgesehen, sie funktionieren zwischen Tabellen allerdings bis dato (Stand Ende 2019) nur auf den Keys selbst (vgl. hier). Mit anderen Worten sind bisher also nur 1:1-Joins und keine 1:N- bzw. N:1-Verknüpfungen über Fremdschlüssel möglich. Bei Joins, die auf Fremdschlüsseln beruhen, wird die rechte Seite Seite immer mit null-Werten aufgefüllt. Naheliegend ist die Vermutung, dass das (noch?) nicht funktioniert, da Kafka bisher keine Indices unterstützt.

Um das näher zu illustrieren benötigen wir noch einige ausgedachte Studenten (mit Studiengängen der HS Mannheim) in einem neuem Kafka-Topic namens "students". Das geht wieder wie oben mit dem Konsole-Producer:

12345:{"vorname":"Klaus","name":"Mayer","matrikelnummer":"12345","studiengang":"UIB"}23456:{"vorname":"Katrin","name":"Müller","matrikelnummer":"23456","studiengang":"IMB"}55555:{"vorname":"Steven","name":"Müller-Mayer","matrikelnummer":"55555","studiengang":"IB"} 

Dieses können wir nun ebenfalls analog zu oben in eine Tabelle wandeln:

create table studis (vorname varchar, name varchar, matrikelnummer varchar, studiengang varchar) with (kafka_topic = 'students', value_format='json', key = 'matrikelnummer');

Nun läuft allerdings folgender Join ins Leere:

select * from exams left join studis on studis.matrikelnummer = exams.matrikelnummer; --> 
1535126105952 | 1 | 1 | Big Data | 1.0 | 12345 | null | null | null | null | null | null
1535126105961 | 5 | 5 | Programmierung 1 | 2.0 | 23456 | null | null | null | null | null | null
    ...

Ein entsprechernder Join einer Tabelle mit einem Stream funktioniert interessanterweise bereits, was  sich wie folgt als Workaround nutzen lässt, um Fremdschlüsselbeziehungen abzubilden:

select * from str_exams left join studis on str_exams.matrikelnummer = studis.matrikelnummer; -->
1535316616713 | 12345 | 1 | Big Data | 1.0 | 12345 | 1535316139962 | 12345 | Klaus | Mayer | 12345 | UIB
1535316616723 | 23456 | 5 | Programmierung 1 | 2.0 | 23456 | 1535316139970 | 23456 | Katrin | Mueller | 23456 | IMB
1535316616723 | 12345 | 17 | Programmierung 1 | 4.0 | 12345 | 1535316139962 | 12345 | Klaus | Mayer | 12345 | UIB

...

Vorausgesetzt, wir haben in unserem Stream keine Wertänderungen auf ein und dem selben Key, die dann natürlich mehrfach gejoint werden würden. Eine weitere Voraussetzung ist, dass beide Datensätze nach der "Join-Spalte" partitioniert sind (sog. Co-Partitioning, d.h., dass die zu joinenden Elemente auf dem gleichen Broker liegen). Das ist üblicherweise auf der Fremdschlüsselseite nicht der Fall, kann aber ggf. mit einem "partition by" erzwungen werden, vgl. hier.

Ein abschließendes Wort der Warnung soll diesen Post beschließen, Kafka ist per default "vergesslich" und löscht Daten nach einer Woche, wenn die Größe eines Topics 1 GB übersteigt. Wer hier tiefer einsteigen möchte, sollte sich die beiden Parameter log.retention.hours und log.retention.bytes genauer betrachten und sie ggf. auf -1 setzen, um das Löschen zu verhindern.

Weiterführend

• Fortsetzung dieses Posts mit einem tieferen Einstieg in die Stream-Verarbeitung
• einführendes Video bei Confluent
• Anwendung von KSQL & Grafana zur Visualisierung von Fahrbewegungen
• Hier noch ein Beispiel wie Tweets ausgewertet werden können