Interaktive Datenverarbeitung mit Apache Zeppelin und Airfield

Die meisten Aufgaben im Bereich Big Data beinhalten Daten-Exploration und –Analyse. Der gemeine Data Scientist will dies machen, ohne jedes Mal die immer gleichen Datenverarbeitungsschritte zu wiederholen, d.h. er will so wenig Zeit wie möglich mit Data Engineering verbringen und so viel wie möglich mit der eigentlichen Analyse. In dieser Hinsicht sind Notebook-Tools wie Jupyter oder Apache Zeppelin unerlässlich, weil sie die Erstellung von Bausteinen ermöglichen, die der Data Scientist in Standardsituationen wiederverwenden kann.

Notebooks sind web-basierte Tools, die datengetriebene, interaktive Datenanalyse sowie kollaborative Dokumente zum Beispiel mit SQL oder Scala machen. Für uns als Big-Data-Entwickler ist die Integration von Zeppelin mit Apache Spark der größte Vorteil, weil wir damit leicht auf Daten aus HDFS, S3 oder einem anderen Datenspeicher zugreifen und Analysen, ETL-Aufgaben und Visualisierungen leicht umsetzen können. Code kann in Scala, Python oder R geschrieben werden, daneben gibt es Spark SQL für den datenbank-affinen Data Scientist. Die Entwicklung von Data Dashboards sowie die Erforschung Deiner Daten sind mit einem Tool wie Zeppelin sehr einfach. Und Du kommst richtig schnell zu Deinen Ergebnissen.

Und wozu brauche ich Airfield?

Zeppelin bietet einen Multi-User Zugang, wodurch Teams zusammenarbeiten können. Das Problem dabei ist, dass sich alle User die Ressourcen teilen müssen, die einer Zeppelin-Instanz zugeteilt sind. Hinzu kommt, dass es je nach Aufgabe sehr spezifische Ressourcenanforderungen geben kann, und die Ressourcen können zudem über eine längere Zeit benötigt werden. Auch benötigen die User teilweise ganz unterschiedliche Bibliotheken für ihre Arbeit.

Mit Airfield kannst Du einen Hub von Zeppelin-Instanzen erstellen, in dem jede Instanz ihre eigenen, spezifischen Ressourcen und Python- oder R-Bibliotheken besitzt. Mitglieder einer Data Science Gruppe, eines Firmenbereichs oder einer Forschungsgruppe können mit ihrer eigenen Zeppelin-Instanz ausgestattet werden – mit genau den Eigenschaften, die sie brauchen. Eine Instanz könnte z.B. GPUs brauchen, während eine andere viele CPUs oder RAM. Mit Airfield kannst Du die Ressourcen von Zeppelin selbst verwalten, sowie die von Spark. Du kannst nicht gebrauchte Instanzen herunterfahren oder löschen. Eine Instanz kann gestartet werden, ohne die anderen zu beeinflussen – zum Beispiel, wenn ein Interpreter hängen bleibt (ein Interpreter ist eine Engine, die Befehle ausführt; es gibt Interpreter für verschiedene Sprachen, z.B. Scala, Python, R oder Spark SQL) – und ohne, dass der Systemadministrator eingreifen muss.

Mit Airfield und Zeppelin loslegen

Um Zeppelin und Airfield besser kennenzulernen, zeigen wir erst, wie man eine neue Zeppelin-Instanz mit Airfield erstellt und dann erstellen wir ein paar einfachen Analysen mit Daten von der New York City Taxi and Limousine Commission. Dabei lernst Du, wie Du in Zeppelin Daten lädst, verarbeitest und darstellst. Den Code dazu findest Du am Ende dieses Artikels.

1. Airfield installieren

Zunächst muss Airfield in deinem DC/OS-Cluster installiert werden. Die konkreten Schritte sind auf Github  dokumentiert.

2. Zeppelin Instanz mit Airfield erstellen

In einem zweiten Schritt kann sich ein Data Scientist mit der Airfield-Benutzeroberfläche eine Zeppelin-Instanz mit maßgeschneiderten Einstellungen erzeugen. Einmal gestartete Instanzen kann er damit auch verwalten.

Als Erstes rufst Du das Airfield UI auf mit dem URL, den Du während der Installation bestimmt hast. Wenn Du das UI aufmachst, bekommst Du einen Überblick über die aktiven Zeppelin-Instanzen. Das allererste Mal ist diese Liste natürlich leer.

Klicke auf den Button ‘Add Instance’ im Hauptfenster, um zu der unteren Eingabemaske zu gelangen.

Wähle einfach den gewünschten Typ, um die Standardkonfiguration zu laden. Du kannst die allgemeinen Einstellungen und Spark Konfiguration editieren und zusätzliche Python- und R-Packages auswählen.

In unserem Beispiel war nur die „requests“ Library von Python notwendig. Um sie zu installieren, wechselst Du auf den Reiter „Libraries“ im Bereich „Instance Settings“ und dort fügst Du „requests“ zu den Python Libraries hinzu. In der Praxis könntest Du Libraries wie Pandas zur Datenvisualisierung oder TensorFlow für Machine Learning brauchen.

3. Zeppelin öffnen

Nachdem die Instanz erzeugt wurde, zeigt Airfield diese in seiner Instanzübersicht an. Neben der Möglichkeit bestehende Instanzen zu starten, stoppen, neu starten oder löschen, kannst Du den URL von einer Zeppelin-Instanz auch aufrufen. Achtung: Wenn Du eine Zeppelin-Instanz löschst, werden alle Notebooks in der Instanz mitgelöscht! Diese solltest du daher vor dem Löschen entsprechend exportieren.

Beim ersten Aufruf eines Zeppelin-URLs siehst Du ein leeres Notebook, wie unten dargestellt.

4. Code schreiben

In Zeppelin kann jeder Absatz (englisch „Paragraph“, d.h. ein Block Code) in seiner eigenen Programmiersprache geschrieben werden. In unserem Beispiel laden wir die Daten aus Amazon S3 mit Python, verarbeiten die Daten mit Scala und zum Schluss verwenden wir SQL, um Daten für die Diagramme bereitzustellen.

Im ersten Schritt werden die NYC Taxi Daten importiert. Die Daten sind auf S3 verfügbar und können daher direkt ins Notebook geladen werden. Dafür verwenden wir Python.

Auf der einen Seite enthalten die Rohdaten mehr Informationen als wir brauchen aber andererseits fehlen einige nützliche Infos. Daher reduzieren und transformieren wir die Daten in einem zweiten Schritt – diesmal mit Scala.

Nachdem wir eine View erstellt haben, können wir die Daten mit SQL abfragen und die Ergebnisse visualisieren. Die Daten für das untere Diagramm wurden mit einem einfachen SQL-Befehl geholt und die gewünschte Darstellung haben wir aus der Menü gewählt: In diesem Fall ein Liniendiagramm mit Anzahl Taxifahrten (y-Achse) pro Stunde (x-Achse) pro Wochentag (eine Linie pro Tag).

Abgekapselte Zeppelin-Instanzen maßgenau erstellen

Mit Airfield können Data Scientists maßgeschneiderte, abgekapselte Zeppelin-Instanzen erstellen und verwalten – ohne die Unterstützung eines Systemadministrators. Data Scientists haben damit ein mächtiges Werkzeug an der Hand, das Ihnen mehr Zeit für coole Datenanalysen lässt.

Wir ermuntern Euch, Airfield zu installieren und auszuprobieren. Wenn Du einen Bug findest oder einen Feature-Wunsch hast, eröffne einfach ein Issue in unserem GitHub Projekt.

---

Code

%python
import requests
response = requests.get("https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2017-07.csv", stream=True)
with open("taxi_data.csv", 'wb') as f:
    for chunk in response.iter_content(1024):
        f.write(chunk)
____________________

%spark
import org.joda.time.DateTime
import org.joda.time.format.DateTimeFormat
import org.apache.spark.sql._
import scala.io.Source
import spark.implicits._
case class TaxiRide(month: Integer, dayOfMonth: Integer, dayOfWeek: Integer, hourOfDay: Integer, duration: Long, distanceKm: Double)
val df = Source.fromFile("/zeppelin/taxi_data.csv").getLines.toList.toDF()
val taxiData = df.filter(line => line.getAs[String](0) != "" && !line.getAs[String](0).startsWith("Vendor")).map(line => {
    val formatter = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")    
    val row = line.getAs[String](0).split(",")
    val startDateTime = DateTime.parse(row(1), formatter)
    val stopDateTime = DateTime.parse(row(2), formatter)
    val month = startDateTime.getMonthOfYear()
    val day = startDateTime.getDayOfMonth()
    val weekday = startDateTime.getDayOfWeek()
    val hour = startDateTime.getHourOfDay()
    val km = row(4).toDouble * 1.60934
    val duration = (stopDateTime.getMillis - startDateTime.getMillis) / 1000
    TaxiRide(month, day, weekday, hour, duration, km)
})
taxiData.createOrReplaceTempView("taxi_data")
taxiData.show()
______________________

%spark.sql
SELECT COUNT(*), dayOfWeek, hourOfDay FROM taxi_data GROUP BY dayOfWeek, hourOfDay
______________________

---

Über die Autoren