Braucht man dafür einen Cluster?
Ob Sie Spark, SQL oder einfaches Python nehmen, hängt davon ab, wie groß die Daten sind. Für kleine Daten reicht oft eine Schleife, für riesige braucht man einen Cluster.
Ein Team will den Umsatz je Land. Eine Person greift zu Spark. Eine andere sagt, eine Python-Schleife mit zehn Zeilen genüge. Beide können recht haben. Heute machen Sie dieselbe Aufgabe auf drei Arten und entscheiden dann mit einem Grund.
Am Ende dieser Lektion
Notiz
Ich kann eine Aufgabe mit einem Spark-DataFrame, mit SQL und mit einfachem Python lösen und sagen, wann was passt.
Schlüsselwörter
| Begriff | Was er bedeutet | Beispiel |
|---|---|---|
| DataFrame | eine Tabelle aus Zeilen und benannten Spalten, auf der Spark arbeitet | |
| Spark SQL | einem DataFrame eine Frage in SQL stellen | |
| Shuffle | Zeilen wandern zwischen Arbeitern, damit gleiche Schlüssel zusammenkommen | Exchange |
| Physischer Plan | die Liste der Schritte, die Spark wirklich ausführt | |
| Im Speicher | klein genug für den Speicher einer Maschine |
Die Zellen, die Lernende ausfüllen, bleiben in dieser Vorschau leer.
Eine Aufgabe, drei Arten
Die Aufgabe ist der Umsatz je Land: Menge mal Preis, für jedes Land addiert. Das Lab hat 24 Bestellungen als orders geladen.
Zuerst als DataFrame-Ausdruck in Spark, dann als Spark-SQL, dann als einfache Python-Schleife über dieselbe Datei. Stimmen die drei nicht überein, irrt sich eine.
Die DataFrame-Art
curl -s -d 'orders.groupBy("country").agg(F.round(F.sum(orders.quantity * orders.unit_price), 2).alias("revenue")).orderBy("country")' 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.
Die SQL-Art
curl -s -d 'SELECT country, round(sum(quantity * unit_price), 2) AS revenue FROM orders GROUP BY country ORDER BY 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 Art mit einfachem Python
curl -s -d 'import csv, collections; c = collections.Counter(); [c.update({r["country"]: float(r["quantity"]) * float(r["unit_price"])}) for r in csv.DictReader(open("/tmp/lab/orders.csv"))]; sorted(c.items())' http://spark:8000/pyHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Dieselbe Aufgabe als Schleife
# Dieselbe Aufgabe als einfache Python-Schleife: der Umsatz je Land.
bestellungen = [["UK", 2, 25.0], ["Japan", 1, 150.0], ["UK", 1, 150.0], ["Japan", 4, 12.5]]
umsatz = {}
for bestellung in bestellungen:
land = bestellung[0]
betrag = bestellung[1] * bestellung[2]
if land in umsatz:
umsatz[land] = umsatz[land] + betrag
else:
umsatz[land] = betrag
print(umsatz)
# Fügen Sie ["France", 3, 12.5] zu den Bestellungen hinzu und starten Sie noch einmal.
Ein DataFrame ist keine Liste
Ein DataFrame sieht aus wie eine Tabelle, durch die man läuft, ist aber ein Plan. Er hält die Zeilen nicht in Ihrem Programm, deshalb scheitern viele Listen-Tricks.
Fragen Sie nach seiner Größe mit count(), einer Aktion. Fragen Sie mit len() nach der Länge, und Python lehnt ab.
Die Länge wie bei einer Liste fragen
curl -s -d 'len(orders)' 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.
Die Größe wie bei Spark fragen
curl -s -d 'orders.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.
Der Plan zeigt, wohin die Daten wandern
Spark kann den physischen Plan ausgeben, den es ausführt. Achten Sie auf das Wort Exchange. Das ist ein Shuffle: Zeilen wandern zwischen Arbeitern, damit gleiche Schlüssel zusammenkommen.
Ein Shuffle ist der teure Teil einer großen Aufgabe. Einen Plan zu lesen sagt, wo er liegt.
Ein physischer Plan, wie Spark ihn ausgibt
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[country#19], functions=[count(1)])
+- Exchange hashpartitioning(country#19, 4), ENSURE_REQUIREMENTS, [plan_id=244]
+- HashAggregate(keys=[country#19], functions=[partial_count(1)])
+- InMemoryTableScan [country#19]
Die wichtigsten Operatoren des Plans auflisten
curl -s -d 'orders.groupBy("country").count()._jdf.queryExecution().simpleString()' http://spark:8000/df | grep -oE "Exchange|HashAggregate|FileScan"Hier 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, Spark sei immer das bessere Werkzeug, weil es das größere ist. Bei 24 Zeilen bringt es nur einen Start und einen Plan. Spark lohnt sich, wenn die Daten nicht mehr auf eine Maschine passen oder die Arbeit verteilt werden muss.
Beispiel: ein Werkzeug wählen
| Die Daten | Eine faire Wahl | Warum |
|---|---|---|
| 24 Zeilen in einer Datei | Eine einfache Python-Schleife | Nichts zu verteilen |
| 2 Millionen Zeilen, die in den Speicher passen | pandas oder SQL in einer Datenbank | Eine Maschine schafft das in Sekunden |
| Milliarden Zeilen auf vielen Maschinen | Spark | Die Arbeit muss verteilt werden |
| Eine Aufgabe, die jede Nacht läuft | Was klar geschrieben und geprüft ist | Jedes Mal dieselbe Antwort |
Die Zellen, die Lernende ausfüllen, bleiben in dieser Vorschau leer.
Weiter: eine andere Frage
Berechnen Sie die verkauften Stück je Produkt mit der DataFrame-Art, der SQL-Art und der Art mit einfachem Python. Stimmen sie überein?
Hinweis: Gruppieren Sie nach product und addieren Sie quantity. Geben Sie dann den Plan Ihrer DataFrame-Zeile aus und finden Sie sein Exchange.