Eine Zahl je Land aus drei Milliarden Zeilen
Jeden Montag will eine Handelskette eine einzige Zahl: den Umsatz je Land. Die Bestelltabelle hat drei Milliarden Zeilen, verteilt auf zwanzig Rechner.
Sie wollen nicht zwanzig Programme schreiben. Sie wollen das Ergebnis einmal beschreiben und das System die Arbeit verteilen lassen.
Am Ende dieser Lektion
Notiz
Ich kann eine Datenverarbeitung als Spark-Pipeline ausdrücken.
Schlüsselwörter
| Begriff | Was er bedeutet | Beispiel |
|---|---|---|
| DataFrame | eine Tabelle mit benannten Spalten, auf der Spark arbeitet | |
| Transformation | ein Schritt, der aus einer Tabelle eine neue macht | filter |
| Aktion | ein Schritt, der den Plan ausführt und eine Antwort liefert | count |
| Lazy | wartet mit der Arbeit, bis eine Antwort verlangt wird | |
| Partition | ein Teil einer Tabelle, den ein einziger Worker bearbeitet | |
| Job | die Arbeit, die Spark für eine Aktion erledigt |
Die Zellen, die Lernende ausfüllen, bleiben in dieser Vorschau leer.
Ein DataFrame ist eine Tabelle, die man abfragt
Bei MapReduce schreiben Sie die Schritte Map und Reduce selbst. Spark lässt Sie das Ergebnis auf einem DataFrame beschreiben: einer Tabelle mit benannten Spalten.
Sie fragen einen DataFrame mit Code ab, etwa mit filter und groupBy, oder mit SQL. Beides ergibt denselben Plan. Das Lab hat eine Datei mit 24 Bestellungen unter dem Namen orders geladen.
Wie das Lab antwortet
Im Terminal des Labs gibt es kein Spark. Es schickt Ihre Fragen per curl an einen kleinen Dienst, der immer mit einer Texttabelle antwortet.
Die Spaltennamen stammen aus der Datei und sind englisch: country, product, quantity. Schicken Sie nur eine Transformation, zeigt der Dienst trotzdem ihr Ergebnis: Er löst dafür selbst eine Aktion aus.
Vier Zeilen ansehen
curl -s -d 'SELECT * FROM orders LIMIT 4' http://spark:8000/sqlHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Die verschiedenen Länder zählen
curl -s -d 'orders.select("country").distinct().count()' http://spark:8000/dfHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Transformationen warten, Aktionen starten
Eine Transformation wie filter oder groupBy erzeugt einen neuen DataFrame, liest aber noch nichts. Spark fügt den Schritt nur seinem Plan hinzu. Man sagt: Spark ist lazy.
Eine Aktion wie count oder collect verlangt eine Antwort. Spark führt dann den ganzen Plan aus. Das ist ein Job.
Beispiel: Schritte und ihre Art
| Schritt | Art | Liest er Daten? |
|---|---|---|
| filter(orders.quantity > 3) | Transformation | Nein, er kommt in den Plan |
| select("country") | Transformation | Nein |
| groupBy("country").sum("quantity") | Transformation | Nein |
| count() | Aktion | Ja, er führt den Plan aus |
| collect() | Aktion | Ja, und holt alle Zeilen zurück |
Die Zellen, die Lernende ausfüllen, bleiben in dieser Vorschau leer.
Ein Plan, der erst auf Verlangen läuft
# Ein Plan ist eine Liste von Schritten. Ein neuer Schritt liest noch keine Daten.
zeilen = [["Frankreich", 2], ["Japan", 1], ["Frankreich", 5]]
plan = []
plan.append(["behalte_mehr_als", 1])
print("Plan:", plan)
def ausfuehren(zeilen, plan):
# Eine Aktion führt den ganzen Plan auf den Zeilen aus.
for schritt in plan:
if schritt[0] == "behalte_mehr_als":
behalten = []
for zeile in zeilen:
if zeile[1] > schritt[1]:
behalten.append(zeile)
zeilen = behalten
return zeilen
print("Ergebnis:", ausfuehren(zeilen, plan))
# Ersetzen Sie die 1 im Plan durch 2 und starten Sie erneut.
Einen Plan ausführen: die großen Bestellungen
curl -s -d 'orders.filter(orders.quantity >= 5).count()' http://spark:8000/dfHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Ein häufiger Fehler
Warnung
Viele denken, eine Zeile mit filter lese die Daten sofort. Das stimmt nicht: Sie fügt dem Plan nur einen Schritt hinzu. Ob die genannten Spalten existieren, prüft Spark trotzdem sofort. Ein falscher Wert in den Daten zeigt sich erst, wenn eine Aktion den Plan ausführt.
Partitionen teilen die Arbeit
Spark teilt einen DataFrame in Partitionen. Jeder Worker, ein Rechenprozess von Spark, addiert seine eigene Partition. Danach addiert Spark die Teilsummen.
Das ist wieder die Idee von Map und Reduce, ohne dass Sie sie schreiben. Die Spark UI zeigt jeden Job und seinen Plan. So sehen Sie, was gelaufen ist.
Teilsummen, dann eine Gesamtsumme
# Drei Partitionen der Mengenspalte, jede bei einem anderen Worker.
partitionen = [[3, 1, 2], [4, 2], [6, 1]]
teilsummen = []
for teil in partitionen:
summe = 0
for menge in teil:
summe = summe + menge
teilsummen.append(summe)
print("Teilsummen:", teilsummen)
gesamt = 0
for wert in teilsummen:
gesamt = gesamt + wert
print("Gesamt:", gesamt)
# Verschieben Sie die 6 in die erste Partition und starten Sie erneut.
Fragen, wie viele Partitionen orders hat
curl -s -d 'orders.rdd.getNumPartitions()' http://spark:8000/dfHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
orders in drei Teile schneiden und zählen
curl -s -d 'orders.repartition(3).rdd.glom().map(len).collect()' http://spark:8000/dfHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Eingegangene Bestellungen je Land
curl -s -d 'SELECT country, count(*) AS bestellungen FROM orders GROUP BY country ORDER BY bestellungen DESC, country' http://spark:8000/sqlHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Die Spark UI zeigt jeden Job
Öffnen Sie die Spark UI über die Leiste des Labs. Der Reiter Jobs listet die Arbeit zu jeder Aktion. Der Reiter SQL / DataFrame listet jede Abfrage mit ihrem Plan.
Starten Sie einen Befehl von oben noch einmal und sehen Sie eine neue Zeile entstehen. So findet man heraus, warum eine Pipeline langsam ist.
Weiterführend: der Durchschnitt je Produkt
Schreiben Sie eine eigene SQL-Abfrage für den Dienst unter /sql. Sie liefert je Produkt die Zahl der Bestellungen und die durchschnittliche Menge je Bestellung, den größten Durchschnitt zuerst.
Sagen Sie vor dem Ausführen voraus, welches Produkt oben steht. Öffnen Sie danach die Spark UI und suchen Sie den Job, den Ihre Abfrage ausgelöst hat.