Logo der codecentric AG, einem in Deutschland führenden IT-Consulting Unternehmen

Logo der codecentric AG einem in Deutschland führenden IT-Consulting Unternehmen

Cloud-Native

Data & Machine Learning

Generative KI

IT-Security

Produktentwicklung

Weitere Leistungen...

Leistungen

Arbeitsweise

Projekte

Berlin

Bielefeld

Dortmund

Erfurt

Frankfurt am Main

Hamburg

Karlsruhe

Leipzig

München

Münster

Nürnberg

Solingen

Stuttgart

Standorte

Blog

Softwerker

Fachmedien

Webinare

Events

Wissens-Hub

Unsere Kultur

Kultur & Karriere Podcast

Deine Entwicklung

Jobs

Karriere

Über uns

Aktuelles

Kontakt

Alle Standorte

Du hast Fragen oder möchtest dich mit mir austauschen? Gern – trag einfach deine Kontaktdaten ein.

Contact Us | Sub-Navi-Kontakt oder Zero Search Hits

Kontakt aufnehmen

Marketing & Community Management

Konntest du keine Einträge zu deinem Thema finden?  

Head of Data

Informatik Aktuell

Data

Data Mesh – ein Gamechanger?

Eine differenzierte Betrachtung des Hype-Themas

Spark Streaming, Flink, Storm, Kafka Streams – das sind nur die populärsten Vertreter einer stetig wachsenden Auswahl zur Verarbeitung von Streaming-Daten in großen Mengen. In diesem Artikel soll es um die wesentlichen Konzepte hinter diesen Frameworks gehen und die drei Apache-Projekte Spark Streaming, Flink und Kafka Streams kurz eingeordnet werden. <h2>Warum Stream Processing?</h2>Die Verarbeitung von Streaming-Daten gewinnt durch die stetig wachsende Anzahl von Datenquellen, die durchgehend Daten produzieren und zur Verfügung stellen, zunehmend an Bedeutung. Neben dem omnipräsenten Internet of Things sind dies zum Beispiel Klickstreams, Daten im Werbegeschäft oder auch Geräte- und Serverlogs. Nun sind unendliche und kontinuierliche Daten kein neues Phänomen. Auch jetzt entsprechen schon viele Daten diesem Schema. Zum Beispiel treten auch Änderungen an Stammdaten kontinuierlich auf, allerdings nur in geringer Frequenz. Stammdaten werden nach dem klassischen Request/Response verarbeitet. Bei zeitunkritischen Änderungen oder größeren Volumen werden die Daten auch gerne gesammelt gespeichert und dann regelmäßig durch Batchprozesse verarbeitet. Diese laufen dann beispielsweise jede Nacht oder auch in kürzeren Intervallen. Tägliche Intervalle reichen aber häufig nicht mehr aus. Gefragt ist Geschwindigkeit: Analysen und Auswertungen werden zeitnah erwartet und nicht Minuten oder gar Stunden später. An dieser Stelle kommt das Stream Processing ins Spiel: Daten werden verarbeitet, sobald sie dem System bekannt sind. Begonnen hat dies mit der Lambda Architektur (vgl. [1]), bei der die Stream- und Batch-Verarbeitung parallel erfolgen, da die Stream-Verarbeitung keine konsistenten Ergebnisse garantieren konnte. Mit den heutigen Systemen ist es auch möglich, nur mit Streaming-Verarbeitung konsistente Ergebnisse nahezu in Echtzeit zu erreichen. (vgl. [2])<h2>Time Matters</h2>Ein wichtiger Aspekt beim Streaming ist die Zeit. Dabei kann im Wesentlichen zwischen drei Zeiten unterschieden werden:<ul><li>Eventzeit: Zeitpunkt, zu dem ein Event tatsächlich auftrat</li><li>Ingestionzeit: Zeitpunkt, zu dem das Event im System beobachtet wurde</li><li>Verarbeitungszeit: Zeitpunkt, zu dem das Event vom System verarbeitet wurde</li></ul><a href="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/doUmMyq0RnMFLNkpcPA4"><img src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/doUmMyq0RnMFLNkpcPA4" alt width="875" height="252" /></a> Abb. 1: Exemplarische Darstellung von Eventzeit und Verarbeitungszeit. Mit verspäteten (Gelb, Grün, Rot) und Out-of-order Events (Blau)In der Praxis ist vor allem Eventzeit im Vergleich zur Ingestion- & Verarbeitungszeit interessant. Die Differenz zwischen der Eventzeit und der Verarbeitungszeit kann stark schwanken. Die Gründe dafür sind vielfältig: Netzwerk-Latenzen, verteilte Systeme, Hardware-Ausfälle oder auch eine unregelmäßige Datenanlieferung. Wenn nach der Verarbeitungszeit verarbeitet wird, ist dies nicht wichtig: Die Daten werden auf Basis der Systemzeit der Verarbeitung analysiert: Wenn ein Event um 12 Uhr eintrifft, ist es irrelevant, dass es bereits um 11 Uhr aufgetreten ist. Der normale Use Case ist dies aber nicht: Wenn ein Event um 11 Uhr auftritt, möchte ich es in der Regel auch zeitlich so betrachten. Die Frage hier ist dann: Wann weiß ich, dass ich alle Events bis 11 Uhr bekommen habe? Wie lange warte ich auf Events? Hier helfen Strategien wie Watermarks, Trigger und Akkumulatoren: <ul><li>Watermarks: Wann habe ich alle Daten zusammen?</li><li>Trigger: Wann soll ich die Berechnung auslösen? </li><li>Akkumulation: Wie füge ich einzelne Berechnungen zusammen, beispielsweise wenn nachträglich Daten folgen?</li></ul>Über diese drei Konzepte ließe sich problemlos ein eigener Artikel schreiben. Tyler Akidau, der Kopf hinter Streaming bei Google, hat dies bereits hervorragend zusammengefasst. Deshalb sei an dieser Stelle für Details sein Artikel empfohlen [3].<h2>State & Window</h2>Jede nicht triviale Anwendung wird eingehende Events miteinander korrelieren. Dafür ist ein Zustand nötig, in dem vorherige Events zwischengespeichert werden. Dieser State kann unendlich gespeichert werden oder explizit zeitlich begrenzt. Ein Beispiel für einen unendlichen gespeicherten State ist eine Lookup-Tabelle mit Metadaten. Ein zeitlich begrenzt State ist beispielsweise ein Window.Bei einem Window werden Daten für einen bestimmten Zeitraum zusammengefasst und analysiert. Dies ist in fast jeder Anwendung nötig, da der Datenstrom ja nie endet. Dabei gibt es verschiedene Typen von Windows:<ul><li>Tumbling Window: nicht überlappende, fixe Zeitabschnitte</li><li>Sliding Window: überlappende, fixe Zeitabschnitte</li><li>Session Window: nicht überlappende Zeitabschnitte unterschiedlicher Länge; definiert durch bestimmte Events oder durch Überschreiten einer bestimmten Zeit zwischen zwei Events</li></ul><img loading="lazy" src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/jMIWsYHTSdqB1gMIVRe5" alt width="760" height="237" />Abb. 2: Tumbling und Sliding Window bei einem Zeitfenster von vier Sekunden und ein Sliding Intervall von zwei Sekunden beim Sliding Window. Innerhalb eines jeden Fensters werden die Werte summiert. <img loading="lazy" src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/m2AyHaoQ7meYWoB9Mhqg" alt width="681" height="139" />Abb. 3: Sessionwindows bei einer Inaktivität von mindestens zwei Minuten zwischen zwei Events für einen Key.Für die Definition von Windows ist die Unterscheidung zwischen Event- und Verarbeitungszeit wichtig: Windows basierend auf Verarbeitungszeit sind sehr einfach zu realisieren, Windows basierend auf Eventzeit benötigen die oben genannten Strategien zur Eventzeit, um nicht unendlich zu wachsen. <h2>API & Laufzeitumgebung</h2>Erste Unterschiede bei den Frameworks lassen sich bei der API und dem generellen Verarbeitungsmodell feststellen. Unterscheiden lässt sich zwischen einem nativen Streaming-Ansatz und dem Microbatching. Beim nativen Streaming werden eingehende Daten direkt verarbeitet wohingegen beim Microbatching die eingehenden Daten zunächst für eine bestimmte Zeit (typischerweise 1 – 30s) gesammelt und anschließend zusammen verarbeitet werden. Der nächste Microbatch kann dann entweder direkt nach dem Abschluss des vorherigen Batches gestartet werden oder erst nach Verstreichen des fixen Intervalls. In beiden Fällen erhöht Microbatching die Latenz, dafür ist das Fehlerhandling etwas einfacher zu realisieren. Der früher häufig genannte Vorteil des sehr hohen Durchsatzes kann heute aber auch von nativen Streaming Frameworks erreicht werden. Zudem bieten diese mehr Flexibilität bei Windows und States.Sichtbar für den Entwickler ist vor allem die API. Auch hier kann zwischen zwei Varianten unterschieden werden: einer komponentenbasierten und einer deklarativen, high-level API. Bei ersterer wird der Fluss durch verschiedene Komponenten beschrieben (Quelle -> Verarbeitung 1 -> Verarbeitung 2 -> Senke), bei letzterer werden die Operationen auf Daten beschrieben ( map, filter, reduce), ähnlich wie bei Scala Collections oder Java 8 Streams. Die Beschreibung von Komponenten bietet mehr Flexibilität bei der Verteilung der Datenströme, während die deklarative API häufig bereits höherwertige Funktion bereitstellt und automatisch Optimierungen vornehmen kann. Zuletzt bleibt noch die Frage: Wo werden die Anwendungen ausgeführt? Auch hier kann man – Überraschung 🙂 – zwei grundsätzliche Alternativen unterscheiden. Einige Frameworks brauchen ein spezielles Cluster bestehend aus Master Nodes und Worker Nodes. Diese Cluster kümmern sich dann auch um das Ressourcenmanagement und Fehlerbehandlung, können dies aber auch auslagern an andere Tools (zum Beispiel YARN oder Mesos). Andere Frameworks kommen als einfache Bibliothek daher, die sich in die eigene Anwendung einbinden lässt. Das Ausführen und Skalieren der Anwendung muss dann von anderen Tools übernommen werden. Hier hat man die volle Flexibilität vom Ausführen eines Jar Files über Docker-Lösungen bis hin zu Mesos & Co.<h2>Verteilte Systeme sind unzuverlässig!</h2>Alle drei Frameworks sind spezialisiert auf die Verarbeitung großer Datenmenge und lösen dies durch horizontale Skalierung. Diese verteilten Systeme sind inhärent unzuverlässig: Einzelne Nodes können ausfallen, das Netzwerk ist inkonstant oder die Datenbank, in der die Ergebnisse geschrieben werden sollen, ist nicht erreichbar. Aus diesem Grund hat jedes Framework unterschiedliche Mechanismen, um bestimmte Garantien zu erreichen. Diese reichen vom Microbatching, bei dem kleine Batches wiederholt werden, über Acknowledges für einzelne Datensätze bis hin zu transaktionalen Updates auf Quelle und Senke. Die erreichten Garantien sind dann in der Regal At-Least-Once, also mindestens einmal verarbeitet, oder Exactly-Once, genau einmal verarbeitet. Da Exactly-Once häufig nur schwierig und mit großem Aufwand zu erreichen ist, sind At-Least-Once-Garantien mit idempotenten Operationen häufig ausreichend sowohl in Bezug auf Geschwindigkeit als auch auf Fehlertoleranz. <h2>Gibt’s da nichts von Apache?</h2>Zeithandling, State & Windows, eine Laufzeitumgebung und das alles in verteilten Systemen: Streaming-Anwendungen sind komplex. Es gibt eine Reihe von Projekten die bei diesen Problemen helfen sollen. Drei davon kurz vorgestellt:Apache Spark (Streaming) Apache Spark ist aktuell eines der populärsten der Projekte im Streaming-Bereich. Gestartet als besseres MapReduce folgte später auch eine Unterstützung für Streaming-Daten. Spark Streaming setzt dabei auf Microbatching mit einer deklarativen API. Aktuell wird dabei nur die Verarbeitungszeit vollständig unterstützt, mit der neuen Structured Streaming API wird seit der Version 2.0 allerdings auch die Unterstützung für Eventzeit-Verarbeitung sukzessive ausgebaut. Das gleiche gilt für die Unterstützung von Windows. Der State wird lokal in Memory oder auf Disk gehalten und per Checkpointing regelmäßig gesichert. Da Spark inzwischen mit jeder Hadoop Distribution ausgeliefert wird, ist die Verbreitung sehr hoch. Ebenso existiert ein großes Ökosystem mit vielen Tools und Konnektoren. Apache Flink Wenn es um Eventzeit-Verarbeitung geht, ist Apache Flink aktuell die erste Wahl. Unterstützt werden Watermarks und Trigger ebenso wie unterschiedliche Window-Operationen. Flink verfolgt dabei einen nativen Streaming-Ansatz und erreicht somit niedrige Latenzen. Ebenso wie bei Spark Streaming wird eine deklarative API genutzt, mit der Möglichkeit sogenannte Rich Functions zu nutzen, in denen beispielsweise ein State genutzt wird. Im Gegensatz zu Spark können verschiedene State-Implementierungen genutzt werden: In-Memory, Festplatte oder RocksDB. Flink ist etwas jünger als Spark, gewinnt aber zunehmend an Verbreitung. Ebenso wachsen die Community und das Ökosystem stetig, sind allerdings noch nicht so groß wie bei Spark. Apache Kafka Streams Das Streaming Framework aus dem Kafka-Ökosystem ist der jüngste Vertreter in dieser Übersicht. Es basiert auf vielen Konzepten, die bereits in Kafka enthalten sind, wie beispielsweise die Skalierung durch Partitionierung der Topics. Auch aus diesem Grund kommt es als leichtgewichtige Bibliothek daher, die in eine Anwendung eingebunden werden kann. Die Anwendung kann dann nach belieben betrieben werden: Standalone, in einem Applikationsserver, als Docker Container oder über einen Resourcen Manager wie Mesos. Flink & Spark hingegen benötigen immer ein Cluster, entweder ein mit den Boardmitteln der Frameworks gebautes oder aber YARN/Mesos. Kafka Streams ist allerdings beschränkt auf Kafka als Quelle und auch als Senke. Die Konnektivität zu anderen Systemen wird dann über Kafka Connect hergestellt. Ansonsten besitzt Kafka Streams neben einer deklarativen auch eine komponentenorientierte API, eine rudimentäre Unterstützung von Eventtime sowie RocksDB als State-Implementierung. Während Kafka selbst schon sehr reif ist und häufig auch in Verbindung mit Flink und Spark genutzt wird, ist die Streaming-Komponente noch recht jung. So ist auch die Community eher klein und die Verbreitung eher gering. Es ist aber zu erwarten, dass beides zeitnah wachsen wird.Update:Kafka Streams nutzt die Konzepte des Beam Models, um den Herausforderungen des Eventzeit Handlings zu begegnen. Streams wird entwickelt auf dem Konzept von <a href="http://docs.confluent.io/3.2.0/streams/concepts.html#duality-of-streams-and-tables">KTables und KStreams</a> , welches genutzt wird, um Eventzeit Verarbeitung zu unterstützen.<h2>Und was passt zu mir?</h2>Nun bleibt zum Schluss die Frage: Welches Framework passt zu mir? Wenn Eventzeit-Verarbeitung benötigt wird, führt aktuell fast kein Weg an Flink vorbei. Ein weiterer Pluspunkt ist die niedrige Latenz. Die wichtigsten Umsysteme (Kafka, Cassandra, Elasticsearch, SQL-Datenbanken) können relativ einfach integriert werden.Die niedrige Latenz und einen einfach zu nutzenden Eventzeit Support ermöglicht auch Kafka Streams. Wenn also Kafka bereits im Einsatz ist <s>und die Verarbeitung eher einfach ist, ohne komplexe Anforderungen an Eventzeit-Verarbeitung</s>, ist Kafka Streams eine gute Alternative. Dafür muss ich hier noch die Umsysteme über Kafka Connect anbinden und mich um die Laufzeitumgebung kümmern. Dies kann aber auch ein Vorteil sein, wenn ich vorhandene Tools, zum Beispiel aus dem Docker-Ökosystem, nutzen kann.Und Spark? Wenn Eventzeit nicht relevant ist und auch Latenzen im Sekundenbereich akzeptabel sind, ist Spark die erste Wahl. Es ist stabil und fast jedes beliebige Umsystem kann einfach eingebunden werden. Außerdem ist es bei Hadoop-Installationen schon vorhanden. Zudem kann der Code, der für Batch-Anwendungen genutzt wird, bei Bedarf auch für die Streaming-Anwendungen verwendet werden, da die API dieselbe ist. Lediglich bei sehr großen States im Terrabyte-Bereich kann es bei Spark zu Problemen kommen. Die Unterstützung für Eventzeit wird mit Spark 2.1 deutlich erweitert.<h2>Fazit</h2>Stream Processing Frameworks vereinfachen die Verarbeitung großer Datenmengen signifikant. Die vorgestellten Frameworks lösen dabei vor allem Probleme im Bereich der verteilten Verarbeitung wodurch einfach zu skalierende Lösungen entwickelt werden können. Ebenso wichtig sind die unterschiedlichen Aspekte der Zeitverarbeitung, die alle Frameworks unterstützen. Hier unterscheiden sich Systeme auch am deutlichsten von Bibliotheken wie Akka Streams, RxJava oder Vert.x. Die vorgestellten Frameworks sind vor allem im Big- und Fast-Data-Bereich angesiedelt, während mit den Bibliotheken auch einfach kleinere reaktive Anwendungen gebaut werden können – dann allerdings in der Regel ohne native Unterstützung für Eventzeit und Clustering.So bleibt festzuhalten, dass die vorgestellten Framework allesamt bei aktuellen Herausforderungen im Fast-Data-Bereich unterstützen können und dabei auch neue Architekturen jenseits der bekannten Lambda-Architektur unterstützen. Dabei ist die Komplexität dieser verteilten System allerdings keinesfalls zu unterschätzen. Dennoch ist davon auszugehen, dass die Verbreitung der Systeme ebenso wie die Funktionalität weiter zunehmen wird. Dieser Artikel erschien zuerst im <a href="https://www.codecentric.de/wissen/softwerker/">Softwerker</a> , dem kostenfreien Magazin der codecentric.<h2>Links</h2>[1] <a href="http://nathanmarz.com/blog/how-to-beat-the-cap-theorem.html" target="_blank">http://nathanmarz.com/blog/how-to-beat-the-cap-theorem.html</a> [2] <a href="https://www.oreilly.com/ideas/questioning-the-lambda-architecture" target="_blank">https://www.oreilly.com/ideas/questioning-the-lambda-architecture</a> [3] https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-102

Verteilte Stream Processing Frameworks für Fast Data & Big Data – Ein Meer an Möglichkeiten

Mit dem neuen Release von Spark 2.1 wurden die Eventzeit-Fähigkeiten von Spark Structured Streaming ausgebaut. Höchste Zeit also den Stand der Unterstützung genauer unter die Lupe zu nehmen und mit Apache Flink – ausgestattet mit einem breiten Support für Event-Time-Verarbeitung – zu vergleichen. In diesem Artikel werde ich beschreiben, wie die drei grundsätzlichen Lösungskonzepte für Eventzeitverarbeitung – Watermarks, Trigger und Akkumulatoren – funktionieren und dann ihre Implementierung in Spark und Flink vergleichen. Basis des Vergleichs ist Spark in Version 2.1 und Flink in Version 1.2.Für einen weitergehenden Vergleich der Frameworks sei auf <a href="/wissens-hub/blog/verteilte-stream-processing-frameworks-fuer-fast-data-big-data-ein-meer-moeglichkeiten">Verteilte Stream Processing Frameworks für Fast Data & Big Data – Ein Meer an Möglichkeiten</a>  verwiesen. Dort werden auch weitere Grundlagen beschrieben, die in diesem Artikel vorausgesetzt werden, beispielsweise zum Stafeful & Window Stream Processing sowie zum Unterschied zwischen Event und Verarbeitungszeit.Zentrales Problem der Eventzeitverarbeitung sind unsortierte sowie verspätete Events. Ereignisse werden nicht exakt zur der Zeit vom System beobachtet zu der sie auftreten. Die Differenz zwischen Eventzeit und Verarbeitungszeit ist dabei nicht konstant. Insbesondere bei zeitlichen Auswertungen erhöht dies die Komplexität des Systems und damit auch der Pogrammierung. Wenn beispielsweise alle Ereignisse zwischen 12 und 13 Uhr verarbeitet werden sollen (eine typische Window Operation), stellen sich mehrere Fragen: Wie soll mit verspäteten Events umgegangen werden? Wie lange soll auf verspätete Einträge gewartet werden? Wie soll das Resultat bei verspäteten Events aktualisiert werden? Google hat mit der Dataflow API, die nun in Apache Beam aufgegangen ist, verschiedene Lösungskonzepte für diese Probleme vorgestellt. Sowohl Spark und Flink implementieren nicht explizit die Dataflow API, allerdings sind die Konzepte ähnlich und können deshalb verglichen werden.Die Dataflow Konzepte sind ein Weg denen Problemen bei der Eventzeit Verarbeitung zu begegnen. Zum Beispiel nutzt Kafka Streams andere Strategien für den Eventzeit Support. Mehr dazu hoffentlich in einem folgenden Blogpost.<h2>Watermarks</h2>Um zu ermitteln, bis zu welcher Zeit Events bereits verarbeitet wurden, gibt es Watermarks. Diese zeigen – wie bei einem Wasserstand –  den bislang erreichten “Zeitpegel” an. Beim Erreichen eines Watermarks wird das Ergebnis der Berechnung dann materialisiert. Wenn beispielsweise der Watermark als der Zeitpunkt des neusten Events abzüglich eines fixen Puffers von 30 Sekunden definiert ist, bedeutet dies “Es wird angenommen, dass nun alle Events bis zum Zeitpunkt x angekommen sind. x ist in diesem Fall der Watermark definiert als Zeitpunkt(neuestes Event) – 30 Sekunden”Zudem gibt es noch heuristische Watermarks, bei denen der Puffer – zum Beispiel basierend auf empirischen Messungen – dynamisch angepasst wird. So könnte das System beobachten, dass Events nachts immer mit deutlicher Verzögerung empfangen werden. Der Puffer könnte auf dieser Grundlage erhöht werden. Ebenso könnte aus der Verzögerung der letzten Stunde die erwartete Verzögerung extrapoliert werden. Außerdem gibt es noch den – zumeist unrealistischen – perfekten Watermark, bei dem die Verarbeitung des Event direkt beim Auftreten des Events stattfindet.Häufig wird im Zusammenhang mit Watermarks auch die “allowed lateness “ genannt – auf Deutsch „die erlaubte Verspätung“. Bei Spark beispielsweise bezeichnet diese den oben beschriebenen Puffer bei den Watermarks. Im Dataflow Model hingegen ist die allowed lateness eine zusätzliche Zeitspanne nach dem Watermark, in der Events nicht ignoriert werden, sondern nachträglich das Ergebnis beeinflussen können. Dazu wird in der Regel über Trigger definiert, wie und wie oft Events verarbeitet werden sollen, die innerhalb der allowed lateness auftreten.Unabhängig von der Interpretation der allowed lateness müssen alle Daten für die Berechnung des Ergebnisses so lange gespeichert bleiben, bis die allowed lateness verstrichen ist, um die Ergebnisse aktualisieren zu können. Anschließend werden die Daten automatisch gelöscht.<h2>Trigger</h2>Während Watermarks den aktuellen Stand der eingegangen Daten anzeigen, lösen Trigger die Berechnung aus. Der Watermark Trigger ist dabei der Einfachste: Dieser wird ausgelöst, sobald der Watermark das Ende des Zeitfenster erreicht – zum Beispiel 13 Uhr bei einem Fenster von 12 bis 13 Uhr. Mit dem oben definierten Watermark mit einem fixen Puffer von 30 Sekunden wird dies um 13:00:30 der Fall sein.Weitere Trigger sind möglich:<ul><li>auf der Verarbeitungszeit, zum Beispiel alle 2 Minuten</li><li>auf der Eventanzahl, beispielsweise alle 100 Events</li><li>bei speziellen Events, zum Beispiel das Ende einer Datei, einem technischen Flush Event oder auch basierend auf dem Inhalt des Events.</li><li>Kombinationen aus den vorherigen dreien</li></ul>Trigger werden typischerweise genutzt, um bereits vor Erreichen des Watermarks Zwischenresultate zu ermitteln oder um die Ergebnisse bei verspäteten Events zu aktualisieren. Möglich wäre es zum Beispiel bei einem Fenster von 10 Minuten, jede Minute zu triggern, dann noch einmal beim Watermark und letztendlich bei jedem verspäteten Event bis die allowed lateness erreicht ist.<h2>Akkumulatoren</h2>Wenn nun das Ergebnis durch die Nutzung von Triggern mehrmals berechnet wird, muss der Entwickler definieren wie mit den einzelnen Teilergebnissen umgegangen wird. Dafür gibt es drei Varianten:<ul><li>Discarding: Bei jedem Trigger wird nur das neue Teilergebnis weitergegeben und die bis hierhin ermittelten Resultate werden anschließend gelöscht.</li><li>Accumulating: Bei jedem Trigger wird das aktuelle Ergebnisse aktualisiert und weitergeben. Das Teilergebnis wird nicht gelöscht.</li><li>Accumulating & Retracting: Wie Accumulating mit der zusätzlichen Information über das vorherige Ergebnis, sodass nachfolgende Operationen ihre Ergebnisse einfach korrigieren können. Dieser Modus ist zwar im DataFlow Model vorgesehen, wird bislang aber von keinem Framework implementiert.</li></ul>Die Wahl des passenden Akkumulators hängt stark von der nachfolgenden Verarbeitung und letztendlichen Senke ab. Wenn diese in der Lage ist Ergebnisse zu aktualisieren, kann der Accumulating Modus genutzt werden. Wenn allerdings jedes Teilergebnis nur einmal als Input für den nächsten Schritt dienen darf, muss der Discarding Mode verwendet werden.<h2>Unterstützung in Spark</h2>Spark benötigt  keinen speziellen Modus zur Eventzeitverarbeitung. Intern unterscheidet sich zunächst nichts von der Verarbeitungszeit. Um in Spark ein Fenster für die Eventzeit zu definieren, muss als Erstes nach dieser gruppiert werden:<div>

```scala
val words = ... // streaming DataFrame mit dem Schema { timestamp: Timestamp, word: String }
// Gruppieren nach Window und Word, berechnen der Anzahl für jede Gruppe
val windowedCounts = words.groupBy(
 window($"timestamp", "10 minutes", "5 minutes"),
 $"word"
).count()
```

</div>Dies unterscheidet sich nicht wesentlich von einem groupBy nach einem Schlüssel nur dass der Schlüssel ein Zeitfenster ist. In diesem Fall ist es ein Zeitfenster von 10 Minuten mit einem Sliding Intervall von 5 Minuten. Zusätzlich können die Einträge noch nach einem fachlichen Key gruppiert werden, in diesem Fall das „word“. Durch das anschließende Zählen erhält man einen WordCount für ein 10 Minuten Fenster.Im obigen Beispiel werden alle Daten unendlich lange gespeichert, damit auch bei verspäteten Events das Ergebnis noch aktualisiert werden kann. Mit Watermarks kann diese Zeit begrenzt werden:<div>

```scala
val windowedCounts = words
 .withWatermark("timestamp", "10 minutes")
 .groupBy(
 window($"timestamp", "10 minutes", "5 minutes"),
 $"word")
 .count()
```

</div>Nun wartet Spark nur 10 Minuten auf verspätete Daten. Anschließend werden die Daten für diese Fenster gelöscht. Aktuell kann nur ein Watermark mit einer fixen erlaubten Verspätung genutzt werden. Bei Spark ist der Watermark aktuell gleichbedeutend zur allowed lateness.Spark implementiert aktuell zwei verschiedene Modi zur Ausgabe des Ergebnisses:<ul><li>Beim Append Mode wird das Ergebnis des Fensters nach Erreichen des Watermarks, hier also 10 Minuten nach dem Ende des Zeitfensters, ausgegeben. Der Watermark fügt hier eine zusätzliche Latenz hinzu.</li><li>Beim Complete Mode wird das bislang errechnete Ergebnis bei jedem Trigger ausgegeben. Allerdings unterstützt Spark bislang nur Trigger auf Basis der Verarbeitungszeit. Zeitunabhängig oder gar zusammengesetzte Trigger werden aktuell noch nicht unterstützt.</li></ul>In Verbindung mit Watermarks kann zudem nur der Append Mode genutzt werden, da für den Complete Mode alle bisherigen Daten vorhanden sein müssen und dementsprechend nicht nach Erreichen des Watermarks gelöscht werden können.Spark bietet somit eine rudimentäre Unterstützung für Watermarks (nur fixe Verspätung) und Trigger (nur auf Verarbeitungszeit). Bei den Akkumulationen entsprechen die implementierten Output Modes von Spark am ehesten dem Accumulating Modus der Datastream API, da immer das komplette Ergebnis für ein Zeitfenster nach Ablauf der erlaubten Verzögerung ermittelt wird.<h2>Unterstützung in Flink</h2>Flink muss zunächst mitgeteilt werden, dass die Verarbeitung nun nach der Eventzeit erfolgen soll. Dies geschieht per<div>

```scala
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTIme);
```

</div>Abhängig von der Zeit-Charakteristik verhalten sich beispielsweise die verschiedenen Window Implementierungen unterschiedlich. Außerdem wird die Eventzeit der Stream Quelle berücksichtigt. Falls die Quelle keine Eventzeit bereitstellt, muss die Eventzeit manuell aus dem Event extrahiert werden mittels Timestamp Assignern. Ebenso muss der Watermark für diese Events festgelegt werden.<div>

```scala
stream.assignTimestampsAndWatermarks(new TimestampAndWatermarkAssigner());
```

</div>Der Entwickler kann den Assigner dabei vollständig selber implementieren oder vordefinierte Implementierungen erweitern. So sind insbesondere für die Watermarks unterschiedliche Implementierungen möglich:<ul><li>mit einem fixen Abstand</li><li>mit einem dynamischen, aber begrenzten Abstand</li><li>oder basierend auf bestimmten Events.</li></ul><div>

```scala
class FixedWatermarkGenerator extends AssignerWithPeriodicWatermarks[SomeEvent] {
 
 override def extractTimestamp(element: SomeEvent, previousElementTimestamp: Long): Long = {
 element.getEventTimestamp
 }
 
 override def getCurrentWatermark(): Watermark = {
 // the watermark is 10s behind the current time
 new Watermark(System.currentTimeMillis() - 10000)
 }
}
```

</div>Der Watermark wird genutzt, um den Zeitpunkt zu bestimmen, an dem die meisten Daten verarbeitet wurden. Zu diesem Zeitpunkt wird dann eine Berechnung durchgeführt. Im obigen Beispiel liegt der Watermark immer 10 Sekunden hinter der aktuellen Zeit. Zusätzlich zum Watermark kann noch eine allowed lateness angegeben werden. Diese bezeichnet die Zeit, die ein Event nach einem Watermark verspätet sein darf. Definiert wird die allowed lateness immer zusammen mit einer Window Operation.<div>

```scala
stream
 .assignTimestampsAndWatermarks(new TimestampAndWatermarkAssigner());
 .keyBy(event -> event.someKey)
 .window((SlidingEventTimeWindows.of(Time.minutes(15), Time.minutes(5)))
 .allowedLateness(Time.minutes(2))
 .apply()
```

</div>Der obige Auszug zeigt ein Sliding Window mit einer Länge von 15 Minuten, einem Slidingintervall von 5 Minuten und einer allowed lateness von 2 Minuten.Wenn es um die Berechnung eines Fensters geht, bietet Flink neben dem Watermark weitere Trigger. Diese Trigger können sowohl auf die Zeit (maßgebend ist die definierte Charakteristik, siehe oben), auf eine bestimmten Anzahl (zum Beispiel alle 100 Events) oder auch auf  eine Mischung aus beiden reagieren. Ebenso können auch dynamisch Trigger registriert werden, die in der Zukunft ausgeführt werden, beispielsweise eine bestimmte Zeit nach einem Event.Bleibt zuletzt noch die Frage nach den Akkumulatoren: Standardmäßig werden die vollständigen aktualisierten Daten an nachfolgende Operationen innerhalb der Streaming Anwendung weitergegeben. Stattdessen kann auch ein Fire and Purge Trigger genutzt werden, bei dem alle aktuellen Daten nach dem Trigger gelöscht werden und alle weiteren Trigger nur die neuen Daten weitergeben. Dies ist dann effektiv ein Discarding Akkumulator.Flink bietet somit mit den unterschiedlichen Watermarks und den flexiblen Triggern sowie Windows umfangreichen Support bei der Eventzeit Verarbeitung. Trotz der Flexibilität ist es ohne größeren Aufwand möglich Standardfälle zu implementieren.<h2>Zusammenfassung und Empfehlung</h2>Es ist offensichtlich, dass Flink schon deutlich länger an der Unterstützung für Eventzeit Verarbeitung arbeitet. So ist es auch zu erklären, dass deutlich mehr Konzepte bereits unterstützt werden. Flink arbeitet zudem kontinuierlich daran, die Dataflow Konzepte vollständig zu unterstützen, beispielsweise mit der <a href="https://issues.apache.org/jira/browse/FLINK-6047">Implementierung des Retracting Akkumulators</a> . Spark hingegen hat erst mit Structured Streaming begonnen die Eventzeitverarbeitung zu unterstützen. Bislang wurden vor allem die Grundlagen geschaffen und mit der Version 2.1 wurde erste Konzepte implementiert. Es ist zu erwarten, dass Spark im Laufe des Jahres die wesentlichen Funktionen implementieren wird. Wer aktuell allerdings Stream Processing mit Eventzeitverarbeitung benötigt, sollte mit Flink starten.

Event-Zeit-Verarbeitung in Apache Spark und Apache Flink

Machine Learning und künstliche Intelligenz sind aktuell in aller Munde und versprechen vielfältige Einsatzmöglichkeiten im Unternehmen. Trotzdem tun sich viele Unternehmen aktuell noch schwer, das Potential der Technologie zu nutzen. „Der Fokus liegt aktuell noch zu sehr auf der Qualität der Machine-Learning-Modelle“, sagt Matthias Niehoff, Head of Data & AI bei codecentric. In der neuen Folge Eine Mate mit berichtet er, was nötig ist, um Mehrwert mit Machine Learning zu schaffen, wie man auch als Mittelständler von den neuen Technologien profitieren kann und welche Entwicklungen in der Zukunft großes Potential haben.
<a href="https://www.youtube.com/watch?v=6pWnLolFxPU" target="_blank"><img src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/PNOkC16jS7eKOCh9aSJF" alt="YouTube" />
zum Video "Data Science & AI | Eine Mate mit Matthias Niehoff #EineMateMit"</a>
<div></div>
 

Eine Mate mit ist ein Format, in dem verschiedenste Menschen aus der IT-Branche aus ihren Projekten und von ihren Erfahrungen erzählen. Ob Speaker auf einem bei der codecentric gehosteten Meetup, ob Kollegen, ob Konferenz-Dauergäste oder das Arbeitstier aus dem letzten Projekt. Die Gemeinsamkeit der Interviews: kurz, knackig, auf den Punkt.<h3>#Eine Mate mit – IT-Wissen kurz und erfrischend</h3>Eine Mate mit möchte Interesse wecken, an Themen und Menschen. Und wer sich danach weiter mit dem Thema beschäftigen möchte, findet weiterführende Links direkt am Ende des Videos und im jeweils zugehörigen Blog-Artikel.Wir freuen uns auf eure Kommentare hier im Blog, auf Twitter unter #EineMateMit oder auf YouTube. Sagt uns eure Meinung zu dem Format, wünscht euch Themen und diskutiert mit. Wir werden in den nächsten Monaten regelmäßig neue Folgen veröffentlichen. Jeweils hier auf dem codecentric-Blog und auf YouTube. 

 <h3>Mehr zum Thema Data, Machine Learning & künstlicher Intelligenz</h3>Interesse am Thema Machine Learning? Dann seht euch doch mal unser <a href="https://www.codecentric.de/wissens-hub/webinare/webinar-data-science-projekte" target="_blank">Webinar</a> zum Thema an! Direkt jetzt abrufbar. Ihr habt Redebedarf? Dann bestellt euch einen „<a href="https://info.codecentric.de/data-lunch" target="_blank">Brown Paper Bag Lunch</a> “ Tech Talk ins Haus – oder auch remote.

Machine Learning in der Praxis. Eine Mate mit … Matthias Niehoff #EineMateMit

Bei der Verarbeitung von Streaming-Daten reichen die Rohdaten aus den Events häufig nicht aus. Meist müssen noch zusätzliche Daten hinzugezogen werden, beispielsweise Metadaten zu einem Sensor, von dem im Event nur die ID mitgeschickt wird.In diesem Blogpost möchte ich auf verschiedene Möglichkeiten eingehen, dieses Problem in Spark Streaming zu lösen. Ich gehe in meinen Beispielen davon aus, dass die zusätzlichen Daten zunächst einmal außerhalb der Streaming-Anwendung liegen und über das Netzwerk – zum Beispiel in einer Datenbank – gelesen werden können. Alle Beispiele und Techniken beziehen sich auf Spark Streaming und nicht auf Spark Structured Streaming. Die wesentlichen Techniken sind<ul><li>Broadcast: für statische Daten</li><li>MapPartitions: für volatile Daten</li><li>MapPartitions + Connection Broadcast: effektives Connectionhandling</li><li>MapWithState: Geschwindigkeit durch einen lokalen State</li></ul><h2>Broadcast</h2>Spark besitzt einen integrierten Broadcast-Mechanismus, mit dem beim Start der Anwendung Daten auf alle Worker Nodes übertragen werden können. Dies hat insbesondere bei großen Datenmengen den Vorteil, dass die Übertragung nur einmal pro Worker Node erfolgt und nicht bei jedem Task.Da die Daten allerdings im weiteren Verlauf nicht aktualisiert werden können, ist dies nur eine Option wenn die Metadaten statisch sind. Das heißt, es dürfen keine zusätzlichen Daten, zum Beispiel Informationen über neue Sensoren, hinzukommen, und es dürfen auch keine Daten verändert werden. Zudem muss das übertragene Objekt serialisierbar sein.In diesem Beispiel soll jeder Sensor-Typ, gespeichert als numerische ID (1,2,…) im Stream Processing durch einen Klartextnamen ersetzt werden (Reifentemperatur, Reifendruck,..). Es wird davon ausgegangen, dass die Zuordnung Typ ID -> Name fix ist. <div>

```scala
val namesForId: Map[Long,String] = Map(1 -> "Wheel-Temperature", 2 -> "Wheel-Pressure")
stream.map (typId => (typId,namesForId(typId)))
```

</div><h6>Ein Lookup ohne Broadcast. Die Map wird für jeden Task serialisiert und auf die Worker Nodes übertragen, auch wenn vorher schon Tasks auf dem Worker ausgeführt wurden.</h6><div>

```scala
val namesForId: Map[Long,String] = Map(1 -> "Wheel-Temperature", 2 -> "Wheel-Pressure")
val namesForIdBroadcast = sc.broadcast(namesForId)
stream.map (typId => (typId,namesForIdBroadcast.value(typId)))
```

</div><h6>Die Map wird über einen Broadcast auf die Worker verteilt und muss anschließend nicht mehr für jeden Task übertragen werden.</h6><h2>MapPartitions</h2>Die erste Möglichkeit nicht statische Daten zu lesen, ist in einer map() Operation. Dabei sollte allerdings nicht map(), sondern mapPartitions() aufgerufen werden. Diese wird nicht für jedes einzelne Element aufgerufen, sondern für jede Partition, die dann mehrere Elemente enthält. Dies ermöglicht es, die Verbindung zur Datenbank nur einmal pro Partition aufzubauen und dann für alle Elemente wieder zu verwenden. Für die Abfrage der Daten gibt es zwei verschiedene Möglichkeiten: Die Nutzung einer Bulk API, um alle Elemente der Partition gemeinsam zu verarbeiten, oder eine asynchrone Variante. Dabei wird für jeden Eintrag eine asynchrone, nicht blockierende Query abgesetzt und die Ergebnisse im Anschluss eingesammelt.<div>

```scala
wikiChanges.mapPartitions(elements => {
 Session session = // create database connection and session
 PreparedStatement preparedStatement = // prepare statement, if supported by database
 elements.map(element => {
 // extract key from element and bind to prepared statement
 BoundStatement boundStatement = preparedStatement.bind(???)
 session.asyncQuery(boundStatement) // returns a Future
 })
 .map(...) //retrieve value from future
})
```

</div><h6>Ein Beispiel für den Lookup von Daten, gespeichert in Cassandra, mit mapPartitions und asynchronen Abfragen</h6>Das obige Beispiel zeigt schematisch einen Lookup mittels mapPartitions: Teure Operationen wie das Herstellen der Verbindung erfolgen nur einmal pro Partition. Für jedes Element wird eine asynchrone, nicht blockierende Query abgesetzt und anschließend die Werte aus den Futures ermittelt. Einige Bibliotheken zum Lesen aus Datenbanken nutzen im Wesentlichen dieses Muster, so beispielsweise das<a href="https://github.com/datastax/spark-cassandra-connector/blob/master/doc/2_loading.md#using-joinwithcassandratable" target="_blank"> joinWithCassandraTable aus dem Spark Cassandra Connector</a> .Warum wird die Connection nicht einmal zu Beginn des Jobs erstellt und dann für jede Partition genutzt? Dafür müsste die Connection serialisiert werden, um dann für jeden Task an die Worker übertragen zu werden. Die Datenmenge dabei wäre in der Tat nicht zu groß, allerdings sind die meisten Connection-Objekte nicht serialisierbar. <h2>Broadcast Connection + MapPartitions</h2>Trotzdem ist es eine gute Idee, die Connection nicht für jede Partition neu aufzubauen, sondern nur einmal pro Worker. Um dies zu erreichen, wird nicht die Connection gebroadcastet, da diese nicht serialisierbar ist (siehe oben), sondern eine Factory, die beim ersten Aufruf die Connection aufbaut und bei allen weiteren aufrufen diese Connection dann zurück gibt. Diese Funktion wird dann in mapPartitions() aufgerufen, um die Verbindung zur Datenbank zu erhalten.In Scala ist es nicht nötig, dafür eine Funktion zu verwenden. Hier kann ein lazy val genutzt werden. Der lazy val wird innerhalb einer Wrapper-Klasse definiert. Diese Klasse ist serialisierbar und kann somit gebroadcastet werden. Auf den Worker Nodes wird dann beim ersten Aufruf eine Instanz der nicht serialisierbaren Connection-Klasse erzeugt.<div>

```scala
class DatabaseConnection extends Serializable {
 lazy val connection: AConnection = {
 // all the stuff to create the connection
 new AConnection(???)
 }
}
val connectionBroadcast = sc.broadcast(new DatabaseConnection)
incomingStream.mapPartitions(elements => {
 val connection = connectionBroadcast.value.connection
 // see above
})
```

</div><h2>mapWithState()</h2>Alle bislang gezeigten Lösungsansätze holen die Daten bei Bedarf aus einer Datenbank. Dies bedeutet in der Regel einen Zugriff über ein Netzwerk für jeden Eintrag oder zumindest für jede Partition. Effizienter wäre es, die Daten direkt in-memory verfügbar zu haben.<a href="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/Lxtm7ADtS2OEXqTmEvEj"><img src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/Lxtm7ADtS2OEXqTmEvEj" alt="Stateful stream procssing within an operation" width="277" height="211" /></a> Spark selbst bietet mit mapWithState() eine Möglichkeit, Daten mittels einen States zu verändern und im Gegenzug ebenfalls den State anzupassen. Der State wird dabei anhand eines Schlüssels verwaltet. Mittels dieses Schlüssels werden die Daten im Cluster verteilt, sodass nicht auf jedem Worker Node alle Daten vorgehalten werden müssen. Ein eingehender Stream muss dementsprechend auch als Schlüssel-Wert-Paar aufgebaut sein.Dieser keyed-State kann auch für einen Lookup genutzt werden. Mittels initialState() kann ein RDD als initialer Zustand übergeben werden. Jegliche Updates können dann allerdings nur noch basierend auf einem Schlüssel durchgeführt werden. Dies gilt ebenso für das Löschen von Einträgen. Es ist nicht möglich, den State komplett zu löschen oder neu zu laden.Um den State zu aktualisieren, müssen zusätzliche Notification Events im Stream vorhanden sein. Diese können beispielsweise aus einem separaten Kafka Topic kommen und mit dem eigentlichen Datenstream zusammengeführt werden (union()). Die gesendete Datenmenge kann dabei von einer einfachen Benachrichtigung mit einer ID, die dann für das Lesen der neuen Daten genutzt wird, bis hin zu dem kompletten neuen Datensatz reichen. Geschrieben wird in das Kafka Topic zum Beispiel wenn Metadaten aktualisiert oder neu angelegt werden. Außerdem können zeitgesteuert Events in das Kafka Topic eingestellt werden oder aber durch einen Custom Receiver in Spark selber erzeugt werden.<a href="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/ickxYXvSeaDhMOnAr61N"><img loading="lazy" src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/ickxYXvSeaDhMOnAr61N" alt="Data Lookup in Spark Streaming using mapWithState" width="644" height="358" /></a> Eine einfache Implementierung kann so aussehen. Zunächst werden hier die Streams von Kafka gelesen und die Schlüssel zusätzlich mit einer Markierung des Datentyps ergänzt (data oder notification). Anschließend werden beide Streams zu einem gemeinsamen Stream vereint und in mapWithState() verarbeitet. Der State wurde zuvor spezifiziert indem die Funktion des States angegeben wurde.<div>

```scala
val kafkaParams = Map("metadata.broker.list" -> brokers)
val notifications = notificationsFromKafka
 .map(entry => ((entry._1, "notification"), entry._2))
val data = dataFromKafka
 .map(entry => ((entry._1, "data"), entry._2))
val lookupState = StateSpec.function(lookupWithState _)
notifications
 .union(data)
 .mapWithState(lookupState)
```

</div>Die lookupWithState Funktion beschreibt die Verarbeitung im State. Folgende Parameter werden dabei übergeben:<ul><li>batchTime: die Startzeit des aktuellen Microbatches</li><li>key: der Schlüssel, in diesem Fall der Originalschlüssel aus dem Stream, zusammen mit der Typmarkierung (data oder notification)</li><li>valueOpt: der Wert zum Schlüssel im Stream</li><li>state: der Wert, der im State zum Schlüssel gespeichert ist</li></ul>Zurückgegeben wird ein Tupel bestehend aus dem Originalschlüssel und dem Originalwert sowie einer zufälligen Zahl, die aus dem State entnommen wird oder – falls noch nicht im State vorhanden – zufällig gewählt wird. <div>

```scala
def lookupWithState(batchTime: Time, key: (String, String), valueOpt: Option[String], state: State[Long]): Option[((String, String), Long)] = {
 key match {
 case (originalKey, "notification") =>
 // retrieve new value from notification or external system
 val newValue = Random.nextLong()
 state.update(newValue)
 None // no downstream processing for notifications
 case (originalKey, "data") =>
 valueOpt.map(value => {
 val stateVal = state.getOption() match {
 // check if there is a state for the key
 case Some(stateValue) => stateValue
 case None =>
 val newValue = Random.nextLong()
 state.update(newValue)
 newValue
 }
 ((originalKey, value), stateVal)
 })
 }
}
```

</div>Zudem kann auch noch der timeout-Mechanismus des mapWithState() genutzt werden, um Events nach einer bestimmten Zeit ohne Aktualisierung aus dem State zu entfernen.<h2>Fazit</h2>Das Laden zusätzlicher Informationen ist ein häufiges Problem in Streaming-Anwendungen. Mit Spark Streaming gibt es eine Reihe von Möglichkeiten, dies zu bewerkstelligen. Am einfachsten ist der Broadcast statischer Daten beim Start der Anwendung. Für volatile Daten ist das Lesen per Partition einfach zu implementieren und bietet bereits eine solide Performance. Bei der Benutzung der Spark States kann die Geschwindigkeit noch gesteigert werden, allerdings ist es insgesamt aufwendiger zu entwickeln.Optimalerweise liegen die Daten immer aktuell direkt auf dem Worker Node vor, auf dem die Daten verarbeitet werden. Dies ist beispielsweise bei der Benutzung des Spark States der Fall. Kafka Streams verfolgt diesen Ansatz noch konsequenter. Hier wird eine Tabelle als Stream behandelt und – vorausgesetzt die Streams sind identisch partitioniert – genauso im Cluster verteilt wie der originale Stream. So sind lokale Lookups möglich.Auch bei Apache Flink wird an effizienten Lookups gearbeitet, hier unter dem Titel <a href="https://cwiki.apache.org/confluence/display/FLINK/FLIP-17%3A+Side+Inputs+for+DataStream+API" target="_blank">Side Inputs</a> .

Datenlookup in Spark Streaming

“Data is the new oil” ist inzwischen ein alter Hut. Jedes Unternehmen versucht, Daten besser zu nutzen, sei es, um die eigenen Prozesse zu optimieren, die Kunden besser zu verstehen oder neue Produkte anzubieten. Dabei stellen fast alle fest: Wir haben viele Daten, wissen aber kaum, wo sie zu finden sind. Insbesondere zwischen den produzierenden und den nutzenden Teams fehlt es an Transparenz. Das hat organisatorische Gründe, erfordert aber auch technologische Lösungen.

Die sozio-technische Idee des Data Mesh hat in den letzten Monaten stark an Popularität gewonnen. Auch hier ist das Ziel der einfache, teamübergreifende Austausch und somit die wertigere Nutzung von Daten im Unternehmen. Das Data Mesh sieht deshalb explizit die “Federated Data Governance” vor. Ziel ist es, die Daten im Unternehmen möglichst nahtlos verwenden und kombinieren zu können. Erreicht werden soll das mit einer Standardisierung – ohne dabei aber die Agilität beim Nutzen und Bereitstellen aus zu bremsen - eine Gratwanderung für Unternehmen auf verschiedenen Ebenen.

Neben organisatorischen Regelungen stehen hier auch Technologien im Fokus. Experten nennen vor allem Datenkataloge, die die Daten katalogisieren, auffindbar machen und zusätzliche Metainformationen anbieten – zum Beispiel zu Ansprechpartner, Qualität oder Abstammung von anderen Daten. Einige Lösungen ermöglichen auch die Abbildung vom Berechtigungsprozess – inklusive des Konzepts eines Datenwarenkorbs, der dann von den Verantwortlichen freigegeben wird. Anschließend geht es ohne Umwege in das Analyse-Tool der Wahl. Zur Auswahl gibt es eine Reihe an Open-Source-Lösungen (z. B. Apache Atlas oder Datahub) und natürlich auch Kaufsoftware. Aber auch in diese Systeme müssen die Informationen erst einmal eingepflegt werden. Bei den Integrationsmöglichkeiten dieser Systeme ist die Entwicklung noch lange nicht abgeschlossen.

Aus technologischer Sicht gibt es also eine Reihe von Werkzeugen, die die Demokratisierung von Daten unterstützen. Wie bei einer agilen oder digitalen Transformation muss die nötige Veränderung zu einem datengetriebenen Unternehmen aber auch in den Köpfen stattfinden.

*Dieser Artikel erschien am 21.10.2022 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr [den Newsletter hier abonnieren](https://www.codecentric.de/#newsletter).*

Data Governance: Wie können wir Daten demokratisieren?

Plattformen sind aktuell ein sehr populäres Konzept, insbesondere in der Softwareentwicklung von Unternehmen. Viele sagen aber auch: So neu ist das doch gar nicht. Wir bieten unseren Entwicklern seit Jahren alle relevanten Tools und Werkzeuge, damit sie sich auf das Wesentliche konzentrieren können: Mehrwerte für Kunden schaffen.

Ist Plattform Engineering also alter Wein in neuen Schläuchen? Ein klares Jein.

Auch klassische IT-Service-Plattformen in Unternehmen versuchen, die nicht differenzierenden, aber notwendigen Arbeiten zu vereinfachen. Bereitgestellt werden CI/CD-Pipelines, Datenbanken, Container-Orchestratoren und Ähnliches. Jedes Team kann sich dann die benötigten Services wie in einem Katalog aussuchen und bekommt sie dann (irgendwann) zur Verfügung gestellt.

Eine interne Entwicklerplattform geht weiter: Verglichen mit den oben beschriebenen IT-Services steht ein Self-Service-Gedanke im Vordergrund und eine einfache, unkomplizierte Nutzung – also kein Ticketsystem mit tagelanger Wartezeit.

Zentral ist auch das Betrachten der Plattform als ein Produkt: Das Angebot richtet sich nach den Bedürfnissen der Nutzer, abgeleitet aus der bisherigen Nutzung und dem Feedback. Sie hilft dem Nutzer das Rad nicht neu zu erfinden, sie nimmt ihm Entscheidungen ab und reduziert so die kognitive Beanspruchung. Und auch wenn viele Nutzer die Plattform nutzen wollen, wird sie nicht zum Bottleneck.
Außerdem ist die Plattform nicht nur eine lose Sammlung von Werkzeugen, sondern ein Mix aus Methoden, Prozessen, Tools und Wissen, der benötigt wird, um schnelle Use Cases umsetzen zu können. Inklusive eines Golden Path, ein klar definierter, schneller, einfacher Weg, bis hin zur produktiven Software – ohne notwendige Compliance-Anforderungen außer Acht zu lassen.

Ein Angebot, das Entwicklungsteam aktiv nachfragen und nicht eines, zu dem sie gezwungen werden. Ein Angebot, das auf die Bedürfnisse der Teams und des Unternehmens zugeschnitten ist und so nicht einfach bei einem der Public-Cloud-Provider verfügbar ist – aber trotzdem auf diesen aufbaut, mit Unterstützung durch Frameworks (z. B. Backstage) und Tools (z. B. Humanitec).

Plattformen bauen schon viele. Aber interne Entwicklerplattformen, als Produkt, mit Fokus auf die Nutzer, mit Golden Path und Self-Service: Das ist immer noch selten.

*Dieser Artikel erschien am 21.10.2022 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr [den Newsletter hier abonnieren](https://www.codecentric.de/#newsletter).*




 Platform Engineering – Machen das nicht alle schon?

Infrastructure as Code (IaC) ist inzwischen ein alter Hut. Frameworks wie Terraform, Ansible und andere haben Standards geschaffen. Kaum jemand provisioniert produktive Systeme heute ohne IaC – sei es in der Cloud oder auf der eigenen Infrastruktur.\nUnd doch ist die Technologie für viele Entwickler oft noch eine Hürde beim Benutzen. Einer der Gründe: Die meisten Tools verwenden eine eigene, deklarative Domain-specific language (DSL), häufig auf Basis von JSON oder YAML oder sogar eine eigene Sprache wie HCL bei Terraform. Schöner ist es doch, die gleiche Sprache zu verwenden, in der auch die eigentliche Software entwickelt wird: TypeScript, Python oder Java.\nDas Cloud Development Kit (CDK) von AWS ermöglicht genau das: Mittels einer verbreiteten Programmiersprache wird die für das Projekt notwendige Infrastruktur beschrieben. Dabei können wir auch die Vorteile einer modernen Sprache nutzen. So ist ein Parameter nicht einfach nur eine Zeichenkette in einer DSL, sondern zum Beispiel ein Wert eines Enum. IDE-Support gibt’s ohne Zusatzaufwand und auch Tools wie Test-Frameworks können einfach verwendet werden.\nEs hat sich zusätzlich ein reichhaltiges Ökosystem gebildet. Unter \nhttps://constructs.dev/\n finden sich eine Vielzahl von Paketen für verschiedene Service-Provider. Mit CDK kann ich dabei auf verschiedenen Granularitäten arbeiten. Von Low-Level Cloud Formation (in das CDK Code bei AWS übersetzt wird) bis hin zu TypeScript-Skripte mit wenigen Zeilen in ein AWS Lambda mit API umwandeln. Und die CDK CLI nimmt mir dabei auch noch Packaging und Deployment ab.\nViele der bereitgestellten Constructs sind natürlich für AWS. Aber mit CDK TF gibt es auch eine Bibliothek, die CDK Code in die Terraform DSL übersetzt und so die Anbindung von Lösungen mit Terraform-Providern ermöglicht. Außerdem gibt es CDK8s mit dem Ziel, CDK für Kubernetes bereitzustellen.\nWeitere spannende Werkzeuge aus dem CDK-Kontext: \nProjen\n mit dem Ziel, nicht nur Infrastruktur, sondern das ganze Entwicklungsprojekt und den Workspace als Code zu beschreiben. \njsii\n ermöglicht CDK in vielen Sprachen zu nutzen – obwohl es intern nur in Typescript geschrieben ist. \nEs ist nicht so, dass für jegliches Provisionieren CDK genutzt werden sollte. Auch mit Terraform lassen sich zum Beispiel Konstrukte wiederverwenden oder parametrisieren. In CDK sind diese Möglichkeiten aber besser in das Framework integriert. \nEin Kritikpunkt, der mehrmals genannt wurde: Entwickler bekommen mit CDK zu viele Freiheitsgrade, was die Wartung des Infrastruktur-Codes massiv verschlechtert. Insbesondere, wenn ein Softwareprodukt vom Entwicklerteam zu einem Betriebs- oder Wartungsteam übergeht, sind klare Grenzen in der Flexibilität durchaus von Vorteil. Wenn auf der anderen Seite ein Entwicklerteam während des gesamten Softwarelebenszyklus für ein Produkt verantwortlich bleibt, sind diese Aspekte ggfs. weniger wichtig. \nZuletzt sei auch noch auf Pulumi hingewiesen, ebenfalls ein IaC-Werkzeug mit ähnlichen Konzepten wie CDK: deklarativ, beschreibt den Zielzustand und nicht den Weg dorthin, Modularisierung und Higher-Level-Abstraktionen. Pulumi ist dabei allerdings nicht auf AWS festgelegt, sondern unterstützt viele weitere Cloud- und SaaS Anbieter. Beim Interpretieren des Code interagiert Pulumi direkt mit den APIs der Cloudanbieter. Bei CDK dient hingegen die AWS CloudFormation als Zwischenschicht. Sowohl Pulumi als auch CDK sind aber eine Weiterentwicklung beim Thema Infrastructure as Code, die eine große Verbreitung finden wird.\nEine Orientierung bei der Auswahl der passenden IaC geben \ndie Kollegen in ihrem Blogbeitrag\n.

AWS Cloud Development Kit – Infrastructure as Code on Steroids

Seit 2017 unterstützen moderne Browser bereits WebAssembly (Wasm), seitdem ist der Hype mal größer, mal kleiner. Aber was ist WebAssembly überhaupt und warum wurde es geschaffen?

WebAssembly ist ein standardisierter Bytecode, der in einer leichtgewichtigen virtuellen Maschine ausgeführt wird, zum Beispiel im Browser. Wer da jetzt direkt an Java, die JVM oder gar Java Applets denkt, ist auf den ersten Blick nicht so falsch unterwegs. Allerdings ist Wasm deutlich leichtgewichtiger, ohne lange Ladezeiten und in einer sicheren Sandbox. Der initiale Fokus auf den Browser legt nahe, dass es ein Ersatz für JavaScript sein soll. Es ist aber vielmehr eine Ergänzung.

Unterstützt wird Wasm vor allem durch die Sprachen C/C++, Rust und Go. Diese Sprachen können direkt in Wasm-Bytecode übersetzt werden. Für andere Sprachen wurde die Runtime nach Wasm übersetzt und sie können so den eigentlichen Code ausführen, z. B. C# oder Python (noch nicht stabil).

2019 wurde mit WASI (WebAssembly System Interface) ein Standard für eine Stand-Alone-Laufzeitumgebung vorgeschlagen, die eine Nutzung auch außerhalb des Browsers erlaubt. Auch die GraalVM unterstützt inzwischen WASI/Wasm. Einige versprechen sich von WASI aufgrund der Portabilität und Geschwindigkeit auch die nächste Stufe bei der Entwicklung von Software für die Cloud.

Als Antwort auf die Ankündigung von WASI twitterte Solomon Hykes, einer der Miterfinder von Docker, dass Docker wohl nicht entstanden wäre, wenn es Wasm bereits gegeben hätte. Im Vergleich zu Docker sieht man allerdings die Schwächen des neuen Standards deutlich: Tooling und User/Developer Experience sind stark ausbaufähig. Auch die Berichte der Early Adopters (unter anderem für die Mediaplayer der BBC und Disney oder auch Google Earth und Figma) sprechen neben den Erfolgen (Geschwindigkeit, Portabilität) immer auch von eigenen Systemen oder Hacks, die hinzu gebaut werden mussten. Mit Wasm ist Stand jetzt auch kein Zugriff auf das DOM einer Webseite möglich. Das muss weiterhin über JavaScript geschehen und diese Kommunikation ist noch hakelig.

Insgesamt ist die Kommunikation rund um WebAssembly sehr von Technik geprägt. Es geht schnell in spannende und herausfordernde Details, es fehlt aber oft der – für die weite Verbreitung notwendige – Überblick über Technologie sowie konkrete Einsatzmöglichkeiten.

Szenarien für den Einsatz gibt es durchaus: Neben den bereits erwähnten Mediaplayern ist das zum Beispiel die Portierung von großen Desktopanwendungen in den Browser, bei der die Logik im Browser ausgeführt wird und um ein modernes Frontend ergänzt wird. Außerdem gibt es Anwendungsfelder im Bereich Gaming, AR/VR sowie Audio- und Videobearbeitung. Im Vergleich zu JavaScript, wo der komplette Code zunächst heruntergeladen und interpretiert werden muss, verspricht Wasm hier durch die bereits erfolgte Kompilierung und Optimierung eine zuverlässige, hohe Performance.

Wasm will mehr als ein Webstandard sein und in Zukunft auch die Cloud ins Visier nehmen. Aus technischer Sicht bringt es dafür bereits vieles mit, aber die Reife des Projekts und des Ökosystems verhindern – aktuell – eine größere Verbreitung jenseits der großen Webanwendungen.

*Dieser Artikel erschien am 02.122022 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr [den Newsletter hier abonnieren](https://www.codecentric.de/#newsletter).*


WebAssembly – Mehr als nur ein Web-Standard 

Rechenleistung und Speicher kosten nicht nur Geld. Sie verbrauchen auch Mengen – potenziell klimaschädlicher – Energie. Das überrascht die Wenigsten, im kollektiven Bewusstsein ist es aber bislang kaum angekommen.

Sehr wohl bewusst ist es natürlich den Public-Cloud-Anbietern selbst. Für sie ist das Thema „Green Cloud“ irgend etwas zwischen bewusstem Fokus auf Nachhaltigkeit, Kosteneffizienz und Marketing. Folglich haben alle großen Anbieter ihre Strategien zum Umgang mit Emissionen, Treibhausgasen und Nachhaltigkeit.

Unterscheiden kann man dabei zwischen den Fragen „Was macht der Anbieter selber?“ und „Was bietet er seinen Nutzern?“. Ein Beispiel: Natürlich ist es super, wenn der Anbieter seine Emissionen um 70 % reduziert. Wenn ich als Unternehmen dann aber Server in der Cloud nur zu 20 % auslaste, habe ich einen ebenso großen Hebel zur Verbesserung. AWS nennt dies explizit „Shared Responsibility“ und beschreibt es als „AWS ist verantwortlich für die Nachhaltigkeit der Cloud“ und „Der Kunde ist verantwortlich für die Nachhaltigkeit in der Cloud“.

Die Anbieter versprechen, in absehbarer Zeit (2025-2030) ausschließlich erneuerbare Energien zu nutzen. Ebenso wird an den Rechenzentren gearbeitet, um beispielsweise eine effizientere Stromnutzung oder eine Weiternutzung der Abwärme zu gewährleisten. AWS hat auf der re:Invent im November gar angekündigt, ab 2030 „Water-positive“ sein zu wollen, also mehr Wasser zurückzugeben als selbst genutzt wird.

Für die Nutzer werden vor allem Dashboards angeboten. Hier kann man sehen, welche Emissionen angefallen sind, grob strukturiert nach geografischen Regionen und genutzten Services. Auch ein Vergleich mit einem „üblichen“ On-Premises Datacenter darf nicht fehlen – schließlich muss der positive Nutzen der Cloud auch hier gezeigt werden. Diese Daten sind immer retrospektiv. Teils gibt es sogar bis zu drei Monate Verzug – eine Tatsache, die zeigt, dass hier bislang nicht so viel investiert wird. Positiv zu erwähnen ist Google, immerhin mit einem „Lowest CO2“-Zeichen bei der Auswahl der Region – und damit bevor die erste Cloud-Ressource erstellt wird.

Sinnvoll wäre es, wenn Nachhaltigkeit bereits beim Entwurf der Lösung in der Cloud eine größere Rolle spielte. Genau so, wie bereits jetzt ganz selbstverständlich Performance und Kosten betrachtet werden (und im späteren Verlauf gerne durch Experteneinsätze weiter optimiert werden). Zumal diese Themen oft Hand in Hand gehen. Hier helfen Blue Prints und Best Practices, effiziente Architekturen und Lösungen, auch unter Berücksichtigung der Umwelt. AWS hat hier mit dem Sustainability Pillar in seinem Well-Architected Framework eine Vorreiterrolle eingenommen.

Unternehmen stehen beim Klimaschutz in der Verantwortung. Die Erwartung ist, dass die Anbieter auf ihrer Seite noch mehr machen, aber ebenso die Kunden in die Lage versetzen, selbst mit wenig Aufwand aktiv zu werden, um so ihre Emissionen zu verwalten und zu reduzieren. Die Cloud-Anbieter arbeiten also daran, aber damit die Cloud wirklich kein Umweltsünder bleibt, müssen Unternehmen das Thema aktiv vorantreiben.

PS: Wenn ihr lernen wollt, wie ihr bei den großen Anbietern eure Emissionen messen könnt, solltet ihr [unseren Blogbeitrag zum Thema lesen](https://www.codecentric.de/wissens-hub/blog/emissionen-cloud-architektur-messen).

*Dieser Artikel erschien am 16.12.2022 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr [den Newsletter hier abonnieren](https://www.codecentric.de/#newsletter).*


Ist die Cloud der große Umweltsünder? 

Sicherheitslücken in Software und Bibliotheken werden immer auftreten, unabhängig davon, wie viel Energie aufgebracht wird, um sie zu vermeiden. An die als Log4Shell bekannte Schwachstelle vor gut einem Jahr werden sich Viele noch schmerzhaft erinnern.

Das Problem: log4j wird in sehr vielen Werkzeugen und Anwendungen eingesetzt. Damit bauen Entwickler wiederum andere Software. Diese Kette von Abhängigkeiten war und ist natürlich unbegrenzt fortsetzbar. Als Folge durchsuchten unzählige Entwicklerteams ihre Software nach Abhängigkeiten zu gefährdeten Bibliotheken. Es gab Unternehmen, die dafür 15 Minuten brauchten – was bei manchen Managern das Gefühl auslöste: Die nehmen das nicht ernst!

Was machen diese Unternehmen anders? Sie haben für ihre Software und SaaS-Angebote Bills of Material (BOMs) erstellt. Materiallisten, mit allen Paketen und Bibliotheken, die diese Lösungen nutzen. Minimal umfassen diese Listen die Bibliotheken inklusive der Version. Besser noch: zu jeder Bibliothek eindeutige IDs, um öffentliche CVE-Listen zu scannen und Informationen über die Lizenz der Bibliothek zu erhalten. Aus diesen Informationen lassen sich dann mit Tools bekannte Schwachstellen inklusive Bewertung und konkreten Maßnahmen zur Behebung anzeigen, zum Beispiel ein Update auf eine andere Version. Diese Informationen stehen dann im Bedarfsfall schnell zur Verfügung.

An welchen Punkten sind die BOMs relevant? Zunächst natürlich während des klassischen CI/CD-Prozesses. Hier werden die Materiallisten erstellt, registriert und bei Bedarf für externe Kunden und Nutzer veröffentlicht. BOMs sind aber keine Black- oder Whitelists. Das Prüfen der BOMs nach dem Erstellen kann beispielsweise auf Ebene eines Artefakt-Repositorys durchgeführt werden, aus dem die Bibliotheken bezogen werden. Eine einmalige Prüfung nur im CI/CD-Prozess reicht natürlich nicht aus. Während des ganzen Lebenszyklus einer Software muss kontinuierlich auf Sicherheitslücken geprüft werden. 

Eine Umsetzung der Prüfung sowohl im Build als auch im Anschluss lässt sich häufig mit einem einfachen, zusätzlichen Schritt im Build und einem Tool wie dependencytrack erreichen. Und doch wird dies selten umgesetzt, da die Priorität auf den Features liegt, die für Nutzer sichtbar sind. 
Gesetzliche Vorgaben aus Brüssel zur Gewährleistung bei digitalen Gütern und damit auch Software sind ein Grund, sich mit dem Thema zu beschäftigen. Außerdem erwarten Softwarekunden ein aktives Risikomanagement von Herstellern: proaktiv, zeitnah und zielgerichtet informieren, wenn Lücken auftreten. Damit hebt man sich von der Konkurrenz ab und steigert das Kundenvertrauen bei verhältnismäßig geringem Kosteneinsatz.

Natürlich gibt es auch Herausforderungen. BOMs schaffen Transparenz. Wenn Lücken gefunden werden, wissen Leute Bescheid, dass sie da sind. Ignorieren ist dann keine Strategie mehr. Das Anti-Pattern „Security by Obscurity” funktionierte aber sowieso noch nie. Ebenso werden manche Lücken schlicht nie behoben. Das heißt, es wird immer ein Grundrauschen von Meldungen geben. Eine zeitnahe Risikoabschätzung der einzelnen Meldungen ist immer essenziell. Das Erkennen selbst hängt stark von der Qualität des Package Managers (die großen wie maven, npm oder pip sind kein Problem) sowie den CVE-Listen ab. Auch bei Containern gibt es (noch) Herausforderungen, wirklich alle Komponenten zu identifizieren, die auf der SBOM landen sollen. Trotzdem sind SBOMs heute aus Security-Sicht wichtiges Handwerkszeug, um bei der stetig wachsenden Anzahl bekannter CVEs zeitnah und zielgerichtet reagieren zu können.

*Dieser Artikel erschien am 06.01.2023 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr [den Newsletter hier abonnieren](https://www.codecentric.de/#newsletter).*

Schneller handeln bei Software-Schwachstellen 

SQL ist weiterhin aus der Datenanalyse nicht wegzudenken – es ist vergleichsweise einfach zu lernen und Anwender können es ohne zusätzliche Werkzeuge auf einer Datenbank ausführen. Entsprechend ist es bei vielen Datenanalysten und Engineers beliebt. Es hat aber auch Schwächen, die besonders im Vergleich zu Python auffallen. Dort gibt es neben den bekannten Machine-Learning-Frameworks auch für Datenpipelines eine Reihe von Werkzeugen, ob zur Datenvalidierung oder Orchestrierung. Dinge, die mit SQL allein gar nicht oder nur mühselig abzubilden sind.

Zudem ist bei Datenplattformen Python oft der Standard für die Verarbeitung von Daten. Da mit SQL aber vergleichsweise gut Transformationen auf Daten ausgedrückt werden können, findet man dann häufig folgendes Pattern: Python-Glue-Code wird geschrieben, der dann SQL-Statements ausführt. Im Extremfall ist das Ganze auch noch über Notebooks verteilt.

[dbt (data build tool) ist ein Kommandozeilentool](https://www.getdbt.com/), um mit SQL solche Datenpipelines strukturiert aufzubauen. Ebenso ermöglicht es die Validierung der Daten direkt in der Pipeline. Es hilft auch bei der Historisierung von Stammdaten, die regelmäßig aus operativen Systemen abgezogen werden. Als Ergebnis gibt es nicht nur saubere und gefüllte Tabellen in einer Datenbank, sondern auch Dokumentation und Abhängigkeitsgraphen.

Mit SQL und dbt lassen sich nicht nur Daten aufbereiten. Auch die darauf folgenden, regelmäßigen Analysen und Auswertungen lassen sich so bequem automatisieren. Inklusive Rückverfolgbarkeit, welche Analyse welche Daten benutzt. Und wenn die Basisdaten inhaltliche Fehler aufweisen, werden die Auswertungen, die darauf basieren, gar nicht erst aktualisiert.

Was muss man dafür tun? Weiterhin SQL schreiben. Außerdem ist ein wenig Konfiguration in YAML notwendig. Optional lassen sich auch Jinja2-Templates verwenden, um komplexere oder wiederkehrende Aufgaben zu lösen. Zudem gibt es eine Reihe von Bibliotheken, die sich unkompliziert einbinden lassen, zum Beispiel für erweiterte Tests auf den Daten. Ausgeführt wird das SQL dann direkt auf der konfigurierten Datenbank. Konzepte für die Trennung von Entwicklung und Produktion sowie die parallele Entwicklung am gleichen Datenmodell sind ebenfalls vorhanden.

dbt ermöglicht es, (Data) Engineering Practices auch für SQL-Datenpipelines anzuwenden. Wer Datenpipelines mit SQL schreibt, sollte es dringend verwenden. Gearbeitet wird aktuell auch an einer Integration von Python. So soll es möglich sein, manche Teile der Pipeline in SQL, andere in Python zu implementieren.  

*Dieser Artikel erschien am 20.01.2023 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr [den Newsletter hier abonnieren](https://www.codecentric.de/#newsletter).*

Bessere SQL-Datenpipelines mit dbt 

Die Frage, was ein Nutzer in einer Anwendung darf, besteht oft aus komplexen Regeln und Konfigurationen, gespeichert in Datenbanken. Regelwerke werden in großen IT-Landschaften in verschiedenen Anwendungen häufig redundant implementiert, teils auch in unterschiedlichen Sprachen. Das führt fast zwangsläufig zu Inkonsistenzen und so zu potenziellen Sicherheitslücken.\nDer Open Policy Agent (OPA) ist eine generische Policy-Engine zum Definieren und Durchsetzen von Regeln. Die Idee ist es, Autorisierungslogik (Policies) und die dazu notwendigen Daten zentral zu verwalten, aber dezentral auszuwerten. Durch einen „Policies as Code“-Ansatz lassen sich technologieneutrale, wiederverwendbare Policies definieren, die einfach ausgewertet und getestet werden können.\nEin konkretes Beispiel: Eine mögliche Autorisierungsanfrage an OPA besteht aus dem aktuellen Benutzer und der Funktion, die aufgerufen werden soll. In den Regeln steht, dass diese Funktion von allen Usern aufgerufen werden darf, die in einer bestimmten Gruppe sind. In
der zusätzlichen Konfiguration steht, in welchen Gruppen ein User ist. Mit den
Informationen aus der Anfrage, den Regeln und der Konfiguration kann OPA eine
Entscheidung treffen.\nIn gängigen Web-Frameworks beschränkt sich die Unterstützung bei der Autorisierung bislang häufig auf die Prüfung von User, Gruppen, Rollen oder Berechtigungen. OPA ist hingegen flexibel in der Ausgestaltung der Regeln. Unterstützt werden zum Beispiel auch Regeln der Art: „Diese Funktion dürfen alle Support-Mitarbeiter ausführen, allerdings nur werktags zwischen 9 und 18 Uhr.“ \nInteressant wird OPA im Zusammenspiel mit der sogenannten Relation Based Access
Control (ReBAC). Hier wird die Berechtigungsprüfung als „Graph Traversal“-Aufgabe gelöst. Ein Beispiel: Die verschiedenen Elemente (Dokumente, Ordner, Ablagen, Gruppen, Benutzer) und Beziehungen (Eigentümer, Bearbeiter, Kommentator) in einem Google Drive werden als Graph modelliert. Wenn ich nun prüfen möchte, ob ein User ein Dokument lesen darf, prüfe ich, ob es einen Pfad durch den Graph vom Dokument zum User gibt (z. B. das Dokument liegt in einem Ordner, der Ordner ist freigegeben für eine Gruppe, der User ist Teil der Gruppe). Falls ja, wird der Zugriff gestattet. Über dieses Konzept lassen sich Prüfungen auch über eine große Zahl von Elementen abbilden, die von Nutzern erstellt werden. \nIn komplexen Anwendungsszenarien mit feingranularer Autorisierungslogik und datengetriebenen Berechtigungsstrukturen ist die Kombination von OPA und einem ReBAC-basierten Modell (z. B. OpenFGA) denkbar. Die Kombination ermöglicht es, die Flexibilität der Regeln in OPA mit den Skalierungsmöglichkeiten der datengetriebenen Autorisierung von ReBAC zusammenzubringen. Mit topaz gibt es auch schon Implementierungen dieser Idee.\nDie zentrale Autorisierung mittels OPA ist vor allem bei verteilten Anwendungen mit unterschiedlichen Stacks interessant. Ebenso bei Software, die einmal entwickelt wird und für unterschiedliche Einsätze unterschiedliche Regeln zur Autorisierung bekommt, zum Beispiel beim Einsatz in anderen Unternehmen. Die Verbindung von OPA mit ReBAC ist ein interessanter Ansatz, insbesondere für Anwendungen mit hohen Anforderungen bezüglich Skalierbarkeit.\nDieser Artikel erschien am 03.02.2023 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr \nden Newsletter hier abonnieren\n.\n\n


Wie Open Policy Agent Entwickler befähigt, Autorisierungen einfach umzusetzen 

Dass die Aufbau- und Ablauforganisation eines Unternehmens wichtig für eine schnelle und flexible IT ist, ist kein Geheimnis. Folglich gibt es eine Reihe von Ansätzen, die hier für Verbesserungen sorgen sollen: agile Ansätze, SAFe und alles, was es rund um das Thema DevOps gibt. Aus dem DevOps-Umfeld kommen auch die Team Topologies, die Manuel Pais und Matthew Skelton 2019 in ihrem gleichnamigen Buch beschreiben.\nSobald eine gewisse Anzahl unterschiedlicher Teams und Interaktionen zwischen ihnen erreicht wurde, ist es sinnvoll, die Teams und Erwartungshaltungen an eben diese klar zu benennen und zu kommunizieren. Ebenso muss die Zusammenarbeit zwischen den Teams geklärt werden und welche Schnittstellen zwischen ihnen genutzt werden sollen. Ziel ist es, dass die Teams ihre Energie in die Lösung der eigenen Herausforderungen stecken, und nicht in die Reibung dazwischen.\nSkelton und Pais haben dafür vier verschiedene Teamtypen definiert: Stream-Aligned, Enabling, Complicated Subsystem und Plattform. Ziel ist es, die crossfunktionalen value-stream-aligned Teams, die eigenständig für ein Produkt zuständig sind, zu unterstützen. Zunächst gibt es da die Enabling Teams, die den Stream-Aligned Teams in Form von Coaching und Mentoring helfen, zum Beispiel zum Thema Security. Sie bauen aber selbst keine Lösungen, die sie betreiben und anbieten.\nDas ist Aufgabe der Plattform-Teams, die aktiv die Infrastruktur vereinfachen. Sie lohnen sich erst, wenn mehrere (5-7) Teams die gleiche Funktionalität benötigen. Dabei sind sie immer auf den Nutzen für den Kunden, die Stream-Aligned Teams, ausgerichtet (s. die Diskussionen rund um das Thema Platform Engineering). Bei Bedarf können dann auch noch Complicated-Subsystem-Teams gebildet werden, die keinen direkten Kundennutzen schaffen, sondern an Problemstellungen arbeiten, die ein tiefes technisches Wissen benötigen. \nNeben den Teamtypen werden auch drei Kollaborationstypen definiert:\n1) Collaboration: Eine enge gemeinsame Zusammenarbeit. \n2) X as a Service: Ein Team bietet eine Dienstleistung als Produkt für die anderen Teams an.\n3) Facilitation: Der Fokus liegt auf dem Enabling eines Teams.\nDie zeitliche Komponente ist dabei wichtig. Zu Beginn kann Collaboration zwischen Teams gewünscht und richtig sein. Mit der Zeit ist eine solche Kollaboration aber eher ein Zeichen für ein organisatorisches Problem. Mittels der Team- und Kollaborationstypen kann man also Teaminteraktionen auch im Laufe der Zeit betrachten und verstehen.\nDie Team Topologies sind für uns eine State-of-the-Art-Methode zum Thema Teamorganisation und werden von uns regelmäßig als Gedankenmodell verwendet. Scrum, Kanban und SAFe fokussieren im Vergleich dazu mehr das Vorgehen und notwendige Rollen. Sie ergänzen die Topologies, insbesondere Kanban passt hervorragend mit den Ideen zusammen. Die Topologies sind sehr fundiert. Hauptsächlich auch, da sie auf Beobachtungen zu DevOps-Antipatterns basieren. Mit „Unfix“ steht schon der nächste Ansatz zum Thema vor der Haustür. Es ist abzuwarten, ob es mehr ist als eine Evolution der bisherigen Herangehensweise. \nDieser Artikel erschien am 17.02.2023 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr \nden Newsletter hier abonnieren\n.

Team Topologies: Ein Gedankenmodell für leistungsstarke Teams 

APIs spielen eine zentrale Rolle bei der Digitalisierung. Extern angeboten, ermöglichen sie das Erschaffen von Ökosystemen und neuen Geschäftsmodellen. Unternehmen wollen gerne selbst als Plattform gesehen werden, auch hier sind APIs unerlässlich. Intern, innerhalb der Unternehmen, dienen APIs nicht nur der Vernetzung unterschiedlicher IT-Systeme, sondern ermöglichen im Optimalfall auch den schnellen, programmatischen Austausch zwischen ganzen Domänen. \nWenn es um das Design von APIs geht, kommt diese Wichtigkeit oft nicht zum Tragen: Die Innensicht der IT-Systeme wird nach außen getragen, es fehlt der Fokus auf die Personengruppe, die die API nutzen wird – und die Frage, welches Problem überhaupt gelöst werden soll, wird nicht beantwortet.\nUnter API Experience versteht man die Anwendung von Consumer-Centric- und UX-Methoden auf das Thema API. Dafür werden zum Beispiel die Ideen des Design-Sprints und Design Thinking adaptiert, um eine API zu schaffen, die den Anforderungen der Konsumenten genügt. Nachdem man das für einige APIs durchgeführt hat, lassen sich daraus grundsätzliche Guidelines für APIs ableiten, spezifisch für den eigenen Kontext. Generische Guidelines können dabei als Grundlage dienen, sollten aber zwingend an die eigene Situation angepasst werden, zum Beispiel an verwendete Architektur- und Technologiekonzepte wie REST vs. Messaging. Konzepte wie Paging, Filtering, der Umgang mit großen Payloads oder langlaufenden Operationen sowie Rate Limiting sind nicht in allen Kontexten sinnvoll. \nBevor man sich konkret über die API und deren Ausgestaltung Gedanken macht, muss geklärt sein, in welcher Rolle man selbst in der API Economy auftritt: Bin ich Anbieter oder vor allem Nutzer? Biete ich meine APIs öffentlich für alle an, nur für eine begrenzte öffentliche Nutzerschaft oder sogar nur intern? Welche Mehrwerte will ich meinen Nutzern anbieten? Hier kommen wir wie schon beim Platform Engineering oder Data Mesh zu dem Punkt, etwas „as a Product“ zu betrachten, in diesem Fall eine API. Themen wie Ownership, Dokumentation und Messen der API, z. B. mittels eines API-Gateway und -Catalog, fließen also auch hier mit ein.\nDer zentrale Teil einer API ist das Datenmodell. Dieses sollte gemeinsam mit den Konsumenten beschrieben werden. Ein möglicherweise bereits vorhandenes Datenmodell, beispielsweise aus einem Data Catalog oder einem Master Data Management, kann dabei als Basis genommen werden. Dieses Vorgehen bedeutet aber nicht, dass zwingend „API First“ gearbeitet werden sollte. Im Gegenteil, für viele Anbieter ist das deutlich schwieriger, da sie aufgrund von Legacy-Systemen, die die Anfragen am Ende beantworten, Einschränkungen haben.\nMit der API Experience sollen der Nutzer und der Nutzen wieder mehr in den Vordergrund gerückt und Organisationen befähigt werden, konsistent und mit wenig Aufwand gute APIs bereitzustellen. Insbesondere für Endpunkte, die von außerhalb der Domäne oder Organisation genutzt werden, eine sinnvolle Sache. \nDieser Artikel erschien am 03.02.2023 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr \nden Newsletter hier abonnieren\n.

Experience: Jetzt auch für APIs 

Crossplane ist ein plattformübergreifendes Kontrollsystem (Control-Plane), das das Management von Cloud-Ressourcen vereinfachen und automatisieren soll. Das Tool ermöglicht es, verschiedene Cloud-Provider und lokale Ressourcen, z. B. Kubernetes-Cluster, als einheitliche Infrastruktur zu verwalten.\nDas Hauptproblem, das Crossplane löst, ist die Herausforderung, verschiedene Cloud-Provider und lokale Ressourcen zu integrieren und zu automatisieren. In einer zunehmend hybriden Cloud-Umgebung ist es schwierig, die Komplexität und Heterogenität der Cloud-Infrastruktur zu managen. Crossplane stellt eine einheitliche API und einheitliche Tools zur Verfügung, um verschiedene Cloud-Provider und lokale Ressourcen effizient zu verwalten. \nRealisiert wird das durch das Verwenden von Composite Resource Definitions (CRD) und damit der Kubernetes-API. Und man ahnt es bereits: Damit das Ganze funktioniert, braucht man ein Kubernetes-Cluster als Management-Cluster (ein minikube ist ausreichend). Ich nutze also Crossplane, um Ressourcen in der Public-Cloud ähnlich zu konfigurieren, wie ich es auch mit Pods und Services in Kubernetes mache. Ist man dadurch, dass man Kubernetes und nicht AWS CloudFormation/CDK, GCP Deployment Manager oder Azure Resource Manager nutzt, auch direkt nicht mehr an den Anbieter gebunden? Wohl kaum, schließlich unterscheiden sich die Services der Cloud-Provider teils deutlich – und auch mit Crossplane werden Services provisioniert, nicht abstrakte Patterns. Das ist ein Ansatz, den beispielsweise dapr.io verfolgt.\nInteressant wird Crossplane so insbesondere in Verbindung mit einer Developer-Plattform. So können die CRDs inklusive Composite und Managed Resources von den Plattform Engineers detailliert konfiguriert und für die Anwendungsentwickler bereitgestellt werden. Diese nutzen dann nur noch die aus dem Kubernetes-Ökosystem bekannten Claims, um die bereitgestellten Ressourcen, zum Beispiel Datenbanken oder Load Balancer, in ihrer Anwendung zu nutzen. Crossplane bietet schon eine Reihe von vordefinierten Ressourcen in Form von Providern, zum Beispiel für die drei großen Cloud-Anbieter. Diese Provider laufen als schlanker Container im Management-Cluster.\nIch nutze also Kubernetes, um Services zu provisionieren. Das klingt nach einem ziemlichen Overkill. Wenn man Kubernetes schon nutzt, kann das Sinn ergeben, da man kein anderes IaC-Tool (z. B. Terraform, Pulumi, CDK) lernen muss. Man bleibt bei der Kubernetes-API, die inzwischen als Industrie-Standard gesehen werden kann. Wenn man Kubernetes noch nicht hat, bekommt man mit Crossplane dann eben jenen Standard statt eine weitere IaC-API. Die Frage, ob das die bessere Alternative ist, muss jeder für sich selbst entscheiden.\nNeben dem Provisionieren von Services in einem Hybrid-Cloud-Szenario wird für Crossplane inzwischen auch der bereits beschriebene Plattform-Engineering-Ansatz verstärkt in den Fokus gesetzt. Aktuell ist die Technologie eher bei den Early Adoptern zu finden und es gibt in der API auch immer nochmal größere, meist non-breaking, Änderungen. Aber mit diesen Änderungen gelingt vielleicht noch eine erfolgreiche Positionierung im Bereich Platform Engineering.\nDieser Artikel erschien am 21.04.2023 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr \nden Newsletter hier abonnieren\n.\n

Crossplane: Eine Lösung für hybride Cloud-Herausforderungen? 

Observability, also die Möglichkeit, das Verhalten von Anwendungen in Echtzeit zu überwachen, Fehler schnell zu identifizieren und Probleme proaktiv anzugehen, ist ein unverzichtbares Element für erfolgreiche digitale Unternehmen. OpenTelemetry ist eine Open-Source-Initiative, die Entwicklern dabei helfen soll, dies in ihren Anwendungen zu implementieren.\nOpenTelemetry ist aus der Zusammenlegung von OpenCensus und OpenTracing entstanden und wird von der Cloud Native Computing Foundation (CNCF) unterstützt. Es stellt eine Reihe an APIs, SDKs und Tools bereit, die es Entwicklern ermöglichen, Metriken, Traces und Logs in Echtzeit zu sammeln und zu transportieren. Durch die Unterstützung einer Vielzahl von Sprachen, Plattformen und Cloud-Anbietern ist es insbesondere für das Distributed Tracing, also dem Einsatz in komplexen, verteilten, heterogenen Landschaften, bestens geeignet.\nDer Hauptvorteil von OpenTelemetry besteht darin, dass Entwicklern eine einheitliche, standardisierte Schnittstelle sowie ein Verfahren für das Sammeln relevanter Informationen in Anwendungen und Systemen bereitgestellt werden. Dies kann sowohl in Form von Agents, als auch mittels SDKs erfolgen. \nDas Zusammenführen, Visualisieren und Auswerten dieser Informationen erfolgt dabei weiterhin in einem der bekannten Backend-Systeme wie Instana, Datadog oder anderen. Die meisten der Anbieter unterstützen dabei inzwischen auch das Sammeln von Informationen mittels OpenTelemetry. Auch in vielen Application-Frameworks sowie -Servern ist OpenTelemetry inzwischen schon von Anbietern integriert.\nMit dem Werkzeug wird die Lücke zwischen Dev und Ops beim Thema Observability zunehmend kleiner. Interessant wird sein, ob die Cloud-Anbieter die Unterstützung für den Standard weiter ausbauen (z. B. eine Integration in Lambdas/Functions) oder sie dort weiter auf die eigenen Angebote setzen. Bei OpenTelemetry liegt der Fokus darauf, die Verbreitung auszubauen, neben den Cloud-Anbietern beispielsweise auch in Form von SDKs für andere Sprachen jenseits von Java/TypeScript. Ebenso ist Observability im Frontend bislang noch wenig standardisiert. \nMöglicherweise gibt es irgendwann auch mal eine Referenzimplementierung für das Zusammenführen und Auswerten, ein Standard ist hier aber nicht zu erwarten. Hier gibt es also noch viele Differenzierungsmöglichkeiten für die Anbieter der diversen Werkzeuge. Nur weil die Daten nun einfacher gesammelt werden können, heißt das nicht, dass auch die Analyse einfach von der Hand geht – und auch nicht, dass die Daten auch tatsächlich gesammelt werden in den Anwendungen. Hier gilt es einen „Golden Path“ vorzugeben, der das Sammeln relevanter Informationen in neuen Systemen von Anfang an berücksichtigt und einfach implementiert. \nDieser Artikel erschien am 26.05.2023 im Rahmen des codecentric-Innovations-Newsletters. Wenn ihr immer direkt über die neusten IT-Trends und -Techniken informiert sein wollt, könnt ihr \nden Newsletter hier abonnieren\n.\n

Zukunftssichere Observability mit OpenTelemetry  

Spark Streaming, Flink, Storm, Kafka Streams –  that are only the most popular candidates of an ever growing range of frameworks for processing streaming data at high scale. This article is about the main concepts behind these frameworks. Furthermore the three Apache projects Spark Streaming, Flink and Kafka Streams are briefly classified.<h2>Why Stream Processing?</h2>The processing of streaming data is gaining in importance due to the steadily growing number of data sources that continuously produce and offer data. In addition to the omnipresent Internet of Things, these include, for example, click streams, data in the advertising business, as well as device and server logs.Infinite and continuous data is not a new phenomenon. Even now, many data correspond to this scheme. For example, changes to master data occur continuously, but only at low frequency. Master data is processed according to the classic request / response pattern. In the case of non-time-critical changes or larger volumes, the data are often stored collectively and then processed regularly by batch processes. These then run, for example, every night or at shorter intervals.However, daily intervals are often not sufficient. Speed is needed: analyzes and evaluations are expected promptly and not minutes or even hours later. At this point stream processing comes into play: data is processed as soon as they are known to the system. This has started with the lambda architecture (cf. [1]), in which the stream and batch processing takes place in parallel, since the stream processing could not guarantee consistent results. With today’s systems, it is also possible to achieve consistent results in almost real-time with streaming processing only (cf. [2]).<h2>Time Matters</h2>An important aspect of streaming is the time. Essentially three different times can be distinguished:<ul><li>Event time: Time at which the event actually occurred</li><li>Ingestion time: Time at which the event was observed in the system</li><li>Processing time: Time at which the event was processed by the system</li></ul><a href="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/doUmMyq0RnMFLNkpcPA4"><img src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/doUmMyq0RnMFLNkpcPA4" alt width="806" height="232" /></a> Abb. 1: Exemplary representation of event time and processing time. With late (yellow, green, red) and out-of-order events (blue)In practice, the event time is particularly interesting compared to the ingestion and processing time. The difference between the event time and the processing time can vary greatly. The reasons are numerous: network latencies, distributed systems, hardware failures or even irregular data delivery. When being processed by the processing time, this is not important. The data is analyzed based on the system time of the processor: if an event arrives at 12 o’clock, it is irrelevant that it has already occurred at 11 o’clock.But this is not the normal use case: If an event occurs at 11 o’clock I would like to treat it in the time it occurred. The question here is: When do I know that I got all the events until 11am? How long do I wait for events? There are several strategies and concepts to solve those problems. On of them is the Dataflow/Beam Model. Here, concepts such as watermarks, triggers and accumulators help:<ul><li>Watermarks: When did I collected all the data?</li><li>Trigger: When should I trigger the calculation?</li><li>Accumulation: How do I merge individual calculations, for example when data is subsequently added.</li></ul>It is easy to write a separate article about these three concepts. Tyler Akidau, the head behind streaming on Google, has already summed this up. Therefore it is recommended to read his article for details [3].<h2>State & Window</h2>Any non-trivial application will correlate incoming events with each other. This requires a state in which previous events are stored temporarily. This state can be stored indefinitely or explicitly limited in time. An example of an infinite stored state is a lookup table with metadata. A temporally limited state is, for example, a window.A window is used to aggregate and analyze data for a specific period of time. This is necessary in almost every application, since the data stream never ends. There are different types of Windows.<ul><li>Tumbling Window: Non-overlapping, fixed time segments</li><li>Sliding Window: Overlapping, fixed time segments</li><li>Session Window: Non-overlapping time segments of different length. Defined by certain events or by exceeding a certain time between two events</li></ul><img loading="lazy" src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/jMIWsYHTSdqB1gMIVRe5" alt width="686" height="214" />Abb. 2: Tumbling and sliding window with a time window of 4 seconds and a sliding interval of 2 seconds with the sliding window. Within each window the values are summed.<img loading="lazy" src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/m2AyHaoQ7meYWoB9Mhqg" alt width="598" height="122" />Abb. 3: Session windows with an inactivity of at least two minutes between two events for a key.For the definition of windows, the distinction between event and processing time is important: windows based on processing time are very simple to implement; windows based on event time need the above event time strategies, in order not to grow infinitely.<h2>API & Runtime Environment</h2>First differences in the frameworks can be found within the API and the general processing model. Differentiating between a native streaming approach and microbatching. In native streaming, incoming data is processed directly while microbatching collects the incoming data for a certain time (typically 1 – 30s) and then processes it together. The next microbatch can then be started either directly after the completion of the previous batch, or only after the fixed interval has elapsed. In both cases, microbatching increases latency, but the handling of errors is somewhat easier. The frequently mentioned advantages of the very high throughput can now also be achieved by native streaming frameworks. They also offer more flexibility for windows and states.Visible to the developer is mainly the API. Here, too, a distinction can be made between two variants: a component-based and a declarative, high-level API. For the former, the flow is described by several components (source -> processing 1 -> processing 2 -> sink), the latter describes the operations on data (map, filter, reduce) similar to Scala Collections or Java 8 streams . The description of components provides more flexibility in the distribution of data streams, while the declarative API often already provides higher-order functions and automatic optimization.Finally, the question is: Where are the applications running? One can distinguish between two – surprise 🙂 – basic alternatives. Some frameworks need a special cluster consisting of master nodes and worker nodes. These clusters then also deal with resource management and error handling, but can also outsource this to other tools (for example, YARN or Mesos). Other frameworks come as a simple library, which can be integrated into your own application. Running and scaling the application must then be taken over by other tools. Here you have the full flexibility from running a jar file via docker up to Mesos or YARN.<h2>Distributed systems are unreliable!</h2>All three frameworks are specialized in processing large amounts of data and solve this by horizontal scaling. These distributed systems are inherently unreliable: single nodes can fail, the network is inconsistent, or the database in which the results are to be written is unavailable.For this reason, each framework has different mechanisms to achieve certain guarantees. These range from microbatching, in which small batches are repeated, via acknowledgments for individual data sets, to transactional updates on source and sink. The guarantees achieved are then usually at-least-once or exactly-once. Since exactly-once is often difficult to achieve, at-least-once guarantees with idempotent operations are often sufficient in terms of both speed and error tolerance.<h2>Isn’t there something that can help us?</h2>Time handling, state & windows, a runtime environment, all in a distributed fashion: streaming applications are complex. There are a number of projects to help with these problems. Three of them briefly presented:Apache Spark (Streaming) Apache Spark is currently one of the most popular projects in the streaming field. Started as a better MapReduce, support for streaming data was added later. Spark streaming relies on microbatching with a declarative API. At the moment, only the processing time is fully supported, but with the new Structured Streaming API the support for event time processing has also been gradually expanded since version 2.0. The same is true for supporting windows. The state is stored locally in memory or on disk and is regularly backed up by checkpointing. Since Spark is now distributed with every major Hadoop distro, the overall distribution is very high. There is also a large ecosystem with many tools and connectors.Apache Flink When it comes to event-time processing, Apache Flink is currently the first choice. Watermarks and triggers are supported as well as different window operations. Flink pursues a native streaming approach and thus achieves low latencies. As with Spark Streaming it offers a declarative API, with the possibility to use so-called rich functions, in which, for example, a state can be utilized. Unlike Spark, the state implementations can be chosen from different implementations: in-memory, hard disk or RocksDB. Flink is slightly younger than Spark, but is gaining in popularity. Likewise the community and the ecosystem is growing steadily, but is not yet as big as with Spark.Apache Kafka Streams The streaming framework from the Kafka ecosystem is the latest representative in this overview. It is based on many concepts already contained in Kafka, such as scaling by partitioning the topics. Also for this reason it comes as a lightweight library, which can be integrated into an application. The application can then be operated as desired: standalone, in an application server, as docker container or via a resource manager such as mesos. Flink & Spark, on the other hand, always need a cluster, either built with the equipment of the frameworks or YARN / mesos. Kafka Streams, however, is limited to Kafka as a source and also as a sink. But you can connect a Kafka topic to other systems through Kafka Connect, with over 60 available connectors. Apart from a declaratory, Kafka also has a component-oriented API, a rudimentary support for event time, and RocksDB as a state implementation. While Kafka is already very mature and often used in connection with Flink and Spark, the streaming component is still quite young. So the community and the spread is rather small. It is, however, to be expected that both will grow rapidly. Update:It should be noted that Kafka Streams does not use the concepts of the Beam Model to tackle the challenges of event time processing. Streams is built on the concept of <a href="http://docs.confluent.io/3.2.0/streams/concepts.html#duality-of-streams-and-tables">KTables and KStreams</a> , which helps them to provide event time processing.<h2>And what suits me?</h2>Finally, the question is: Which framework suits me?  If event-time processing is required and you do not mind working with the concepts of the Beam Model, you could go with Apache Flink. Another advantage is the low latency. The most important systems (Kafka, Cassandra, Elasticsearch, SQL databases) can be integrated relatively easily.The low latency and an easy to use event time support also apply to Kafka streams. So if Kafka is already in use <s>and the processing is rather simple, without complex requirements for event processing</s> (Streams can also be used for more complex stream processing), Kafka Streams is a good alternative. For this you have to connect the other systems, like databases, via Kafka Connect and care about the runtime environment. This can also be an advantage if I can use existing tools, for example from the Docker ecosystem.And Spark? If event time is not relevant and latencies in the seconds range are acceptable, Spark is the first choice. It is stable and almost any type of system can be easily integrated. In addition it comes with every Hadoop distribution. Furthermore the code used for batch applications can also be used for the streaming applications as the API is the same.Only with very large states Spark can cause problems. The support for event time is expanded with Spark 2.1.<h2>Conclusion</h2>Stream processing frameworks significantly simplify the processing of large amounts of data. The presented frameworks primarily solve problems in the area of distributed processing, whereby easy-to-scale solutions can be developed. Equally important are the different aspects of the time processing, which all frameworks support in some way.That is what  distinguishes those systems from libraries such as Akka Streams, RxJava, or Vert.x. The presented frameworks are mainly located in the Big and Fast Data area, while the libraries can also be used to build smaller, more reactive applications, but usually without native support for event time and clustering.It remains to be noted that the presented framework can all help with current challenges in the fast data area and also support new architectures beyond the well-known lambda architecture. However, the complexity of these distributed systems is in no way to be underestimated. Nevertheless, it is to be assumed that the spread of the systems as well as the functionality will continue to grow.<h2>Links</h2>[1] <a href="http://nathanmarz.com/blog/how-to-beat-the-cap-theorem.html">http://nathanmarz.com/blog/how-to-beat-the-cap-theorem.html</a> [2] <a href="https://www.oreilly.com/ideas/questioning-the-lambda-architecture">https://www.oreilly.com/ideas/questioning-the-lambda-architecture</a> [3] <a href="https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-102">https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-102</a>

Distributed Stream Processing Frameworks for Fast & Big Data

With the new release of Spark 2.1, the event-time capabilities of Spark Structured Streaming have been expanded. It is time to take a closer look at the state of support and compare it with Apache Flink – which comes with a broad support for event time processing. . In this article, I will describe how three basic solutions for event processing – watermarks, triggers and accumulators – work and then compare their implementation in Spark and Flink. Base of the comparison is Spark in version 2.1 and Flink in version 1.2.For a broader comparison of the frameworks refer to <a href="/wissens-hub/blog/distributed-stream-processing-frameworks-fast-big-data" target="_blank">Distributed Stream Processing Frameworks for Fast & Big Data</a> . The linked article also describes further basics, which are presupposed for this comparison, for example stafeful & window stream processing as well as the difference between event and processing time.The central problems of event processing are unsorted as well as delayed events. Events are not observed by the system exactly at the time they occur. The difference between the event time and the processing time is not constant. Especially in the case of temporal evaluations, this increases the complexity of the system and thus also the coding. If, for example, all events between 12 and 13 o’clock are to be processed (a typical window operation), multiple question arise: How to deal with late events? How long to wait for delayed entries? How to update the result in case of delayed entries? Google has presented various solutions for these problems with the Dataflow API, which has now been released in Apache Beam. Both Spark and Flink do not explicitly follow the Dataflow API, but the concepts are similar and therefore can be compared.The concepts used in the Dataflow API are one way to tackle event-time processing problems. Kafka Streams for example takes other strategies for event-time support. More on this hopefully in a following blog post.<h2>Watermarks</h2>To determine up to which time events have already been processed, there are watermarks. These show – as with a water level – the so far reached “time level”. When a watermark is reached, the result of the calculation is materialized. For example, if the watermark is defined as the time of the latest event minus a fixed buffer of 30 seconds, it means “It is assumed that now all events have arrived until the time x”. x is in this case the watermark defined as time(newest event) – 30 seconds.There are also heuristic watermarks where the buffer is dynamically adapted, for example based on empirical measurements. Thus the system could observe that events are always received at night with a clear delay. The buffer could be increased on this basis. Likewise, the expected delay could be extrapolated from the delay of the last hour. In addition, there is also the – mostly unrealistic – perfect watermark, where the event is processed directly when it occurs.In the context of watermarks, the “allowed lateness” is often mentioned. For Spark, for example, this is the watermark buffer described above. In the Dataflow Model, however, allowed lateness is an additional period after the watermark, in which events are not ignored, but can subsequently influence the result. Therefore rules are defined – for example by means of triggers – how and how often events are to be processed that occur within allowed lateness.Regardless of the interpretation of allowed lateness, all data for the calculation of the result must remain stored until the allowed lateness has elapsed in order to update the results. The data is then automatically deleted.<h2>Trigger</h2>While watermarks display the current state of the data received, triggers materialize the calculation. The watermark trigger is the easiest way to do this: it is fired as soon as the watermark reaches the end of the window – for example, 1:00 pm with a window from 12 am to 1 pm. With the watermark defined above with a fixed buffer of 30 seconds, this will be the case at 13:00:30.Other triggers are possible:<ul><li>On the processing time, for example, every 2 minutes</li><li>On the event number, for example every 100 events</li><li>At special events, for example the end of a file, a technical flush event or based on the content of the event.</li><li>A combination of the three above</li></ul>Triggers are typically used to determine intermediate results before the watermark is reached, or to update the results for delayed events. It would be possible, for example, with a window of 10 minutes to trigger every minute, then again at the watermark and finally at each delayed event until the allowed lateness is reached.<h2>Accumulation</h2>If the result is calculated more than once by the use of triggers, the developer must define how the individual partial results are handled. There are three variants:<ul><li>Discarding: With each trigger, only the new partial results are passed on and the results obtained up to this point are subsequently deleted.</li><li>Accumulating: The current results are updated and passed on each trigger. The results are not deleted.</li><li>Accumulating & Retracting: Like accumulating with the additional information about the previous result so that subsequent operations can easily correct their results. This mode is provided in the DataFlow model, but is not yet implemented by any framework.</li></ul>The choice of the appropriate accumulation strategy depends strongly on the subsequent processing and the final sink. If this is able to update results, the accumulating mode can be used. If, however, every partial result must only serve once as input for the next step, the discarding mode must be used.<h2>Support in Apache Spark Streaming</h2>Spark does not need a special mode for event processing. Internally, nothing is different from the processing time. To define a window for the event time in Spark, you must first group it by the window<div>

```scala
val words = ... // streaming DataFrame with schema { timestamp: Timestamp, word: String }
 
// Group by window and word, calculate the count for each group
val windowedCounts = words.groupBy(
 window($"timestamp", "10 minutes", "5 minutes"),
 $"word"
).count()
```

</div>This is not significantly different from a groupBy on a key, but with a time window as key. In this case it is a window of 10 minutes length with a sliding interval of 5 minutes. In addition, the entries can still be grouped by a non-technical key, in this case the “word”. With the count at the end you get a word count for a 10 minutes window.In the example above, all data is stored indefinitely so that the result can be updated even in the case of delayed events. With watermarks this time can be limited:<div>

```scala
val windowedCounts = words
 .withWatermark("timestamp", "10 minutes")
 .groupBy(
 window($"timestamp", "10 minutes", "5 minutes"),
 $"word")
 .count()
```

</div>Now spark only waits 10 minutes for delayed data. The data for this window is then deleted. Currently only a watermark with a fixed allowed delay can be used. At Spark, the watermark is currently equal to the allowed lateness.Spark currently implements two different modes to output the result:<ul><li>In the case of the append mode, the result of the window is output after reaching the watermark, ie 10 minutes after the end of the time window. Early triggers, ie when reaching the end of the window, are not possible. The watermark adds an additional latency.</li><li>In the case of complete mode, the previously calculated result is output with every trigger. However, Spark currently only supports triggers based on the processing time. Time-independent or even composite triggers are currently not supported.</li></ul>In conjunction with watermarks, only the append mode can be used, since all the existing data must be available for the complete mode and can not be deleted after reaching the watermark.Spark thus offers a rudimentary support for watermarks (only fixed delay) and triggers (only on processing time). For the accumulations, the implemented output modes of Spark most likely correspond to the accumulating mode of the DataFlow API, since the complete result for a time window is always determined after expiration of the allowed delay.<h2>Support in Apache Flink</h2>It must first be indicated to Flink that the processing is to take place on the event time. This is done by<div>

```scala
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTIme);
```

</div>Depending on the time characteristic, for example, the different window implementations behave differently. In addition, the event time of the stream source is taken into account. If the source does not provide an event time, the event time must be manually extracted from the event using timestamp assigners. The watermark must also be defined for these events.<div>

```scala
stream.assignTimestampsAndWatermarks(new TimestampAndWatermarkAssigner());
```

</div>The developer can completely implement the assigner himself or extend predefined implementations. In particular, different implementations are possible for watermarks:<ul><li>With a fixed distance</li><li>With a dynamic but limited distance</li><li>Or based on specific events.</li></ul><div>

```scala
class FixedWatermarkGenerator extends AssignerWithPeriodicWatermarks[SomeEvent] {
 
 override def extractTimestamp(element: SomeEvent, previousElementTimestamp: Long): Long = {
 element.getEventTimestamp
 }
 
 override def getCurrentWatermark(): Watermark = {
 // the watermark is 10s behind the current time
 new Watermark(System.currentTimeMillis() - 10000)
 }
}
```

</div>The watermark is used to determine the time when most of data for a window was processed. At this time, a calculation is executed. In addition to the watermark, an allowed lateness can be specified. This is the period of time that an event may be delayed beyond the watermark. The allowed lateness is always defined in conjunction with a window operation..<div>

```scala
stream
 .assignTimestampsAndWatermarks(new TimestampAndWatermarkAssigner());
 .keyBy(event -> event.someKey)
 .window((SlidingEventTimeWindows.of(Time.minutes(15), Time.minutes(5)))
 .allowedLateness(Time.minutes(2))
 .apply()
```

</div>A sliding window with a length of 15 minutes, a sliding interval of 5 minutes and an allowed lateness of 2 minutes.When it comes to calculating a window, Flink offers additional triggers besides the watermark. These triggers can react to a certain number (for example all 100 events), to the time (either event or processing time) or to a mixture of both.. It is also possible to dynamically register triggers, which are executed in the future, for example a certain time after an event.Last but not least the question about the accumulators: By default the complete updated data is passed on to subsequent operations within the streaming application. Instead, a fire and purge trigger can be used, in which all current data is deleted after the trigger and all other triggers only pass on the new data. This is then effectively a discarding accumulator.As a result, Flink offers extensive support during event processing with its various watermarks and flexible triggers as well as windows. Despite the flexibility, it is possible to implement standard cases without major effort.<h2>Summary and recommendation</h2>It is obvious that Flink has been working much longer on support for event processing. So it is to be explained that significantly more concepts are already supported. In addition, Flink continuously works on supporting the dataflow concepts, for example with the <a href="https://issues.apache.org/jira/browse/FLINK-6047">implementation of the retractable accumulator</a> . Spark, on the other hand, has only started to support event processing with Structured Streaming. So far the basic principles have been created and the first concepts have been implemented with version 2.1. It is to be expected that Spark will implement the essential functions in the course of the year. But if you need stream processing with event processing now, you should start with Flink.

Event time processing in Apache Spark and Apache Flink

When processing streaming data, the raw data from the events are often not sufficient. Additional data must be added in most cases, for example metadata for a sensor, of which only the ID is sent in the event.In this blog post I would like to discuss various ways to solve this problem in Spark Streaming. The examples assume that the additional data is initially outside the streaming application and can be read over the network – for example in a database. All samples and techniques refer to Spark Streaming and not to Spark Structured Streaming. The main techniques are<ul><li>broadcast: static data</li><li>mapPartitions: for volatile data</li><li>mapPartitions + connection broadcast: effective connection handling</li><li>mapWithState: speed up by a local state</li></ul><h2>Broadcast</h2>Spark has an integrated broadcasting mechanism that can be used to transfer data to all worker nodes when the application is started. This has the advantage, in particular with large amounts of data, that the transfer takes place only once per worker node and not with each task.However, because the data can not be updated later, this is only an option if the metadata is static. This means that no additional data, for example information about new sensors, may be added, and no data may be changed. In addition, the transferred objects must be serializable.In this example, each sensor type, stored as a numerical ID (1,2, …), is to be replaced by a plain-text name in the stream processing (tire temperature, tire pressure, ..). It is assumed that the assignment type ID -> name is fixed.<div>

```scala
val namesForId: Map[Long,String] = Map(1 -> "Wheel-Temperature", 2 -> "Wheel-Pressure")
stream.map (typId => (typId,namesForId(typId)))
```

</div><h6>A lookup without broadcast. The map is serialized for each task and transferred to the worker nodes, even if tasks were previously executed on the worker.</h6><div>

```scala
val namesForId: Map[Long,String] = Map(1 -> "Wheel-Temperature", 2 -> "Wheel-Pressure")
val namesForIdBroadcast = sc.broadcast(namesForId)
stream.map (typId => (typId,namesForIdBroadcast.value(typId)))
```

</div><h6>The map is distributed to the workers via a broadcast and no longer has to be transferred for each task.</h6><h2>MapPartitions</h2>The first way to read non-static data is in a map() operation. However, not map() should be used but mapPartitions(). mapPartitions() is not called for every single element, but for each partition, which then contains several elements. This allows to connect to the database only once per partition and then to reuse the connection for all elements.There are two different ways to query the data: Use a bulk API to process all elements of the partition together, or an asynchronous variant: an asynchronous, non-blocking query is issued for each entry and the results are then collected.<div>

```scala
wikiChanges.mapPartitions(elements => {
 Session session = // create database connection and session
 PreparedStatement preparedStatement = // prepare statement, if supported by database
 elements.map(element => {
 // extract key from element and bind to prepared statement
 BoundStatement boundStatement = preparedStatement.bind(???)
 session.asyncQuery(boundStatement) // returns a Future
 })
 .map(...) //retrieve value from future
})
```

</div><h6>An example for an lookup on data stored in Cassandra using mapPartitions and asynchronous queries </h6>The above example shows a lookup using mapPartitions: expensive operations like opening the connection are only done once per partition. An asynchronous, non-blocking query is issued for each element, and then the values are determined from the futures. Some libraries for reading from databases mainly use this pattern, such as the <a href="https://github.com/datastax/spark-cassandra-connector/blob/master/doc/2_loading.md#using-joinwithcassandratable" target="_blank">joinWithCassandraTable from the Spark Cassandra Connector</a> .Why is the connection not created at the beginning of the job and then used for each partition? For this purpose, the connection would have to be serialized and then transferred to the workers for each task. The amount of data would not be too large, but most connection objects are not serializable.<h2>Broadcast Connection + MapPartitions</h2>However, it is a good idea not to rebuild the connection for each partition, but only once per worker node. To achieve this, the connection is not broadcasted because it is not serializable (see above), but instead a factory that builds the connection on the first call and then returns this connection on all other calls. This function is then called in mapPartitions() to get the connection to the database.In Scala it is not necessary to use a function for this. Here a lazy val can be used. The lazy val is defined within a wrapper class. This class can be serialized and broadcasted. On the first call, an instance of the non-serializable connection class is created on the worker node and then returned for every subsequent call.<div>

```scala
class DatabaseConnection extends Serializable {
 lazy val connection: AConnection = {
 // all the stuff to create the connection
 new AConnection(???)
 }
}
val connectionBroadcast = sc.broadcast(new DatabaseConnection)
incomingStream.mapPartitions(elements => {
 val connection = connectionBroadcast.value.connection
 // see above
})
```

</div><h6>A connection creation object is broadcasted and then used to retrieve the actual connection on the worker node.</h6><h2>MapWithState()</h2>All solution approaches shown so far retrieve the data from a database, if necessary. This usually means a network call for each entry or at least for each partition. It would be more efficient to have the data directly in-memory available.<a href="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/Lxtm7ADtS2OEXqTmEvEj"><img src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/Lxtm7ADtS2OEXqTmEvEj" alt="Stateful stream procssing within an operation" width="277" height="211" /></a> With mapWithState() Spark itself offers a way to change data by means of a state and, in turn, also to adjust the state. The state is managed by a key. This key is used to distribute the data in the cluster, so that all data must not be kept on each worker node. An incoming stream must therefore also be constructed as a key-value pair.This keyed state can also be used for a lookup. By means of initialState(), an RDD can be passed as an initial state. However, any updates can only be performed based on a key. This also applies to deleting entries. It is not possible to completely delete or reload the state.To update the state, additional notification events must be present in the stream. These can, for example, come from a separate Kafka topic and must be merged with the actual data stream (union()). The amount of data sent, can range from a simple notification with an ID, which is then used to read the new data, to the complete new data set.Messages are published to the Kafka topic, for example, if metadata is updated or newly created. In addition, timed events can be published to the Kafka topic or can be generated by a custom receiver in Spark itself.<a href="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/ickxYXvSeaDhMOnAr61N"><img loading="lazy" src="https://eu-central-1.graphassets.com/AiE4QoWSSiIQO3k152ugkz/ickxYXvSeaDhMOnAr61N" alt="Data Lookup in Spark Streaming using mapWithState" width="644" height="358" /></a> A simple implementation can look like this. First, the Kafka topics are read and the keys are additionally supplemented with a marker for the data type (data or notification). Then, both streams are merged into a common stream and processed in mapWithState(). The state was previously specified by passing the function of the state to the StateSpec.<div>

```scala
val kafkaParams = Map("metadata.broker.list" -> brokers)
val notifications = notificationsFromKafka
 .map(entry => ((entry._1, "notification"), entry._2))
val data = dataFromKafka
 .map(entry => ((entry._1, "data"), entry._2))
val lookupState = StateSpec.function(lookupWithState _)
notifications
 .union(data)
 .mapWithState(lookupState)
```

</div>The lookupWithState function describes the processing in the state. The following parameters are passed:<ul><li>batchTime: the start time of the current microbatch</li><li>key: the key, in this case the original key from the stream, together with the type marker (data or notification)</li><li>valueOpt: the value to the key in the stream</li><li>state: the value stored in the state for the key</li></ul>A tuple consisting of the original key and the original value as well as a number will be returned. The number is taken from the state or – if not already present in the state – is chosen randomly.<div>

```scala
def lookupWithState(batchTime: Time, key: (String, String), valueOpt: Option[String], state: State[Long]): Option[((String, String), Long)] = {
 key match {
 case (originalKey, "notification") =>
 // retrieve new value from notification or external system
 val newValue = Random.nextLong()
 state.update(newValue)
 None // no downstream processing for notifications
 case (originalKey, "data") =>
 valueOpt.map(value => {
 val stateVal = state.getOption() match {
 // check if there is a state for the key
 case Some(stateValue) => stateValue
 case None =>
 val newValue = Random.nextLong()
 state.update(newValue)
 newValue
 }
 ((originalKey, value), stateVal)
 })
 }
}
```

</div>In addition, the timeout mechanism of the mapWithState() can also be used to remove events after a certain time without updating from the state.<h2>Conclusion</h2>Loading additional information is a common problem in streaming applications. With Spark Streaming, there are a number of ways to accomplish this.The easiest way is to broadcast static data at the start of the application. For volatile data, read per partition is easy to implement and provides a solid performance. With the use of the Spark states, the speed can be increased further, but it is more complex to develop.Optimally, the data is always directly present on the worker node, on which the data is processed. This is the case, for example, with the use of Spark states. Kafka streams pursue this approach even more consistently. Here, a table is treated as a stream and – provided the streams are identical partitioned – distributed in the same way as the original stream. This makes local lookups possible.Apache Flink is also working on efficient lookups, here under the title<a href="https://cwiki.apache.org/confluence/display/FLINK/FLIP-17%3A+Side+Inputs+for+DataStream+API"> Side Inputs</a> .

Lookup additional data in Spark Streaming

Databricks is a great platform when it comes to data management and governance, mostly due to the unity catalog. But Spark as an engine for processing the data is just ok'ish, especially when data is not really big. New engines like polars, datafusion or duckdb are better suited for this and provide interesting options.

Sure, you can run anything you want in Databricks Notebooks and workflows by just installing the library. But the interesting part is to access the data stored in the Databricks unity catalog.

There are effectively two ways to query the data with duckdb inside Databricks:

- the obvious one: read the data from the unity catalog table with Spark, convert it to pandas/arrow and use duckdb from there
- the direct one: read the delta file directly from the underlying storage using duckdb, without spark in between

For the impatient: you can find the code for both ways in this [gist on github](https://gist.github.com/mniehoff/4120f3874559097bbc1a01bd4f0993bb). This also includes all the imports which I mostly omitted in the examples below for the sake of brevity. Also, you need to install duckdb e.g. by `%pip install duckdb` in a notebook cell.

## Read with pyspark, process with duckdb
We use standard pyspark to read the data and convert it to arrow format afterwards. Then we can query it using duckdb.
```python
import duckdb

# read the data with spark
spark_df = spark.read.table("samples.nyctaxi.trips")

# create an arrow table from the spark dataframe that can be queried using duckdb
# the name of the variable (here `trips`) is the name of the table in duckdb. 
# duckdb discovers this automatically.
trips = spark_df.toArrow()

duckdb.sql("SELECT COUNT (*) FROM trips")
```
`toArrow()` is a new method added in Spark 4. Spark 4 is not released yet, but Databricks regularly adds new (unreleased) features from the open-source version to their runtime.

The method works like the `toPandas()` method, but instead of returning a pandas DataFrame, it returns an Arrow Table. Also the same limitations apply: like with `toPandas()` all data is loaded into the memory of the driver node. Therefore this will not work for datasets larger than memory. Arrow is a bit more efficient than pandas in terms of memory usage, so it might work for a bit larger datasets than with pandas. It will definitely be faster in terms of performance for the conversion. So you should prefer arrow over pandas, although duckdb could also query pandas dataframes.

When you want to convert the duckdb result back to a Spark dataframe, e.g. for using the built-in Databricks visualization you can pass an arrow table to `SparkSession.createDataFrame()`, starting with Databricks Runtime 16:

```python
result = duckdb.sql(sql).arrow()
display(
   spark.createDataFrame(result)
)
```
In older runtimes you could use Pandas instead of arrow as an intermediate format.

## Read delta files directly with duckdb

While the above way works it has some limitations. Most notably:
- It loads all data into the memory of the driver node
- Reading with Spark and converting to arrow adds a performance penalty

duckdb is actually pretty good at reading “larger than memory” data directly from cloud blob storages. So it would be nice to use these functionalities.

To do so, we use the Databricks Unity Catalog feature of [temporary table credentials](https://learn.microsoft.com/en-us/azure/databricks/external-access/credential-vending). This is [an API](https://docs.databricks.com/api/azure/workspace/temporarytablecredentials/generatetemporarytablecredentials) that provides the URL to the delta files in the blob storage and a temporary token valid for 1 hour to read this data (In Azure it's an SAS token).

In order to use this API:
- you must authenticate against the API
- the principal you authenticate with must have `EXTERNAL_USE_SCHEMA` permissions

```python
w = WorkspaceClient(
   host=spark.conf.get(
       "spark.databricks.workspaceUrl"
   )
)

def get_temporary_credentials_for_table(table: str):
   table_id = w.tables.get(table).table_id
   return w.temporary_table_credentials.generate_temporary_table_credentials(table_id=table_id, operation=TableOperation.READ)

cred = get_temporary_credentials_for_table("samples.nyctaxi.trips")
```

The credential API does not accept a full qualified table name, but only a table ID which we have to retrieve first. Be aware that you might need to upgrade the databricks-sdk on your cluster to the latest version, as the temporary credentials API is quite new.

After we have retrieved the credential, we can store it in duckdb. 

*Note: setting azure_transport_option_type to curl is needed, as otherwise duckdb struggles to handle certificates correctly on Databricks.*

```python
storage_account_name = re.search('@(.*).dfs.', cred.url).group(1)
sql = f"""
SET azure_transport_option_type = 'curl';
CREATE OR REPLACE SECRET (
   TYPE AZURE,
   CONNECTION_STRING 'AccountName={storage_account_name};SharedAccessSignature={cred.azure_user_delegation_sas.sas_token}'
);
"""
duckdb.sql(sql)
```
And finally query the files using duckdb:
```python
sql = f"""
SELECT * FROM delta_scan("{cred.url}")
"""
duckdb.sql(sql).show()
```

It is also possible to use duckdb outside of Databricks to process data stored in the Unity Catalog. You would use the same approach as above (getting the credentials using the Databricks API, store it in duckdb, do the query). But as you call the API from outside of databricks, the metastore must be [enabled for external data access](https://learn.microsoft.com/en-us/azure/databricks/external-access/admin#enable-external-data-access-on-the-metastore). And of course, the blob storage must be reachable network-wise from wherever you want to do the query.

## Closing remarks
The temporary table credentials API is a really nice addition to the unity catalog. In my opinion it absolutely needed for databricks, so that other engines can be used as well. Sticking to spark as the only processing option would otherwise prevent some modern, efficient processing techniques. Maybe we get an even smoother integration of duckdb and the databricks managed unity catalog, something along the lines of [duck/uc_catalog](https://github.com/duckdb/uc_catalog), which is a PoC for the open source unity catalog with [some Azure Databricks support](https://github.com/duckdb/uc_catalog/issues/7) already built-in.

Access Databricks UnityCatalog from duckdb

Contact Us | Fragen an Mitarbeiter

Sprich mich an

Möchtest du dich über Data Mesh austauschen?

BLOG

codecentric Blog

Auf unserem Blog findest du viele Artikel zu Tech-Themen von unseren Mitarbeitenden. Vielleicht ist auch hier ein Thema für dich dabei. Schau gerne mal rein!

WISSENS-HUB

Events – codecentric mittendrin

Community

Softwareentwicklung

Kultur

Der Softwerker

Hier gibt's noch mehr spannende Infos von uns für euch

Vielen Dank für deine Anfrage. Ich melde mich bei dir.

Vielen Dank für deine Anmeldung. Eine E-Mail zur Bestätigung ist zu dir unterwegs.

Diese Seite ist eine Vorschau. \nKlicken Sie hier\n, um den Vorschaumodus zu verlassen.

Zu deiner Suche haben wir leider keine passenden Blogartikel gefunden. 

Du hast noch Fragen zu diesem Thema? Dann sprich mich einfach an.

Du hast Fragen oder möchtest dich mit mir austauschen? Gern - trag einfach deine Kontaktdaten ein.

Die \nDatenschutzerklärung\n der codecentric AG habe ich zur Kenntnis genommen und bestätige das mit der Absendung des Formulars.*

Entdecke spannende weiterführende Themen und lass dich von der codecentric Welt inspirieren.

Ich möchte eure Inhalte abonnieren. Diese Einwilligung kann ich jederzeit widerrufen bzw. den Blog Newsletter abbestellen. Die Datenschutzerklärung habe ich gelesen. Ich bin damit einverstanden, dass die codecentric AG, Hochstr. 11, 42697 Solingen, mich über eigene Dienstleistungen, branchenrelevante Events, Webinare und Publikationen der codecentric AG über die von mir angegebene E-Mail-Adresse informiert. Meine Einwilligung ist freiwillig. Ich kann meine Einwilligung jederzeit mit Wirkung für die Zukunft widerrufen (z.B. über den hierfür vorgesehenen Abmeldelink in unserem Newsletter, per E-Mail an datenschutz@codecentric.de oder per Post an die oben genannte Adresse der codecentric AG) erklären. Weitere Hinweise zur Verarbeitung meiner personenbezogenen Daten und zu meinen Rechten als betroffene Person finde ich hier: \nhttps://www.codecentric.de/datenschutzerklaerung\n

Bitte wähle deine/n gewünschte/n Standort/e aus.

Wir möchten dich als Mensch kennenlernen. Erzähle uns, wer du bist, worauf du Wert legst, wie du gerne arbeitest und was du von uns als Arbeitgeber erwartest.

Bitte füge deine Bewerbungsunterlagen ​(mind. Lebenslauf, gerne auch mehr)​​ ​als eine ​PDF-​Datei ​hinzu.

Es gilt die Datenschutzerklärung der codecentric AG, abrufbar unter \nDatenschutzbestimmungen\n. Mit dem Absenden des Formulars bestätige ich, dass ich diese zur Kenntnis genommen habe.

Job application form field placeholder 10

Keine passende Rolle gefunden? Bewirb dich bitte trotzdem initiativ! Jenseits von starren Schubladen zählen für uns vielmehr Motivation, Kompetenz und der Cultural Fit.

Job application form field placeholder 11

Dies ist ein Pflichtfeld, bitte fülle es aus.

Data Mesh – ein Gamechanger?

Aus dem Inhalt:

Hier gibt's noch mehr spannende Infos von uns für euch

codecentric Blog

Events – codecentric mittendrin

Der Softwerker

Jetzt für unseren Newsletter anmelden