Einführung
Ein Einzelhändler erhält Bestelldatensätze und möchte daraus nützliche Informationen gewinnen. Bevor Sie Verkaufsberichte berechnen, benötigen Sie eine Möglichkeit, Zeilen darzustellen, ihre Spalten zu beschreiben und das Verarbeitungswerkzeug zur Ausführung aufzufordern. Sie beginnen mit drei Datensätzen, die klein genug sind, um sie mit bloßem Auge zu prüfen.
Big Data bezeichnet Datenaufgaben, deren Umfang, Eingangsgeschwindigkeit oder Verarbeitungsbedarf einen bequemen Ansatz auf einem einzelnen Rechner übersteigen kann. Apache Spark ist eine Datenverarbeitungsengine, die Arbeit auf Rechner eines Clusters verteilen kann. PySpark ist ihre Python-Schnittstelle. Ein DataFrame stellt Zeilen mit benannten, typisierten Spalten dar und unterstützt Operationen wie Filtern und Aggregieren. Spark verarbeitet Daten; ein separates Datei- oder Speichersystem liefert dauerhafte Ein- und Ausgaben. Dieser Kurs beginnt mit winzigen Beispielen, damit Sie vor der Skalierung die Korrektheit nachvollziehen können.
Sie erstellen eine SparkSession, zeigen einen DataFrame an, prüfen Schema und Zeilenanzahl und fügen vor einem erneuten Lauf einen Datensatz hinzu. Die vorbereitete Umgebung verwendet local[2]: Spark läuft auf diesem einen Rechner mit zwei Ausführungsthreads. Damit lernen Sie echte Spark-APIs; eine verteilte Bereitstellung ist ein separates Thema. Sie benötigen grundlegende Python-Variablen, Listen und Funktionsaufrufe; der Python-Lernpfad hilft bei diesen Voraussetzungen. pandas, Hadoop, Hive, Java, Installation oder Dateien früherer Labs sind nicht erforderlich. Verwenden Sie das Terminal.
Führen Sie ein vollständiges kleines DataFrame-Skript aus
In diesem Schritt führen Sie ein vollständiges Skript aus, das drei Python-Datensätze in eine Spark-Tabelle umwandelt. Beginnen Sie in Ihrem Arbeitsverzeichnis:
cd /home/labex/project
Eine SparkSession ist der Einstiegspunkt zum Erstellen von DataFrames und für den Zugriff auf Sparks APIs für strukturierte Daten. Der Builder legt einen Anwendungsnamen fest, wählt die lokale Ausführung und erstellt die Sitzung. local[2] erlaubt zwei Ausführungsthreads auf diesem Rechner; es erzeugt keine zwei Clusterrechner. setLogLevel reduziert spätere Informationsmeldungen, damit die Lern-Ausgabe leichter zu finden ist.
Im Python-Quelltext macht from pyspark.sql import SparkSession diese installierte Klasse für Ihr Programm verfügbar. Ein Punkt greift auf ein benanntes Mitglied zu, und Klammern rufen eine Methode auf: df.show() fordert diesen DataFrame auf, seine Zeilen anzuzeigen. Der Builder verkettet mehrere Aufrufe; die umgebenden Klammern erlauben einen Ausdruck über mehrere Zeilen. Lesen Sie jede Zeile als eine Einstellung, bevor getOrCreate() die Sitzung beschafft.
Die Liste orders enthält drei Tupel, also in Klammern gesetzte Wertegruppen. Jedes Tupel liefert eine Zeile in derselben Reihenfolge wie die deklarierten Spalten. Die Schemazeichenfolge benennt jede Spalte und ihren Typ: Textkennungen und Produkte, gefolgt von einer ganzzahligen Menge. createDataFrame verwendet diese Datensätze und das Schema. show fordert Spark zur Zeilenanzeige auf, und stop gibt die Sitzung nach Abschluss frei.
Speichern Sie das vollständige Skript. Das Shell-Zeichen > schreibt die benannte Datei; der Marker PYTHON in Anführungszeichen beginnt ein Here-Dokument, das den eingeschlossenen Python-Text wörtlich erhält. Kopieren Sie bis einschließlich der letzten PYTHON-Zeile:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
spark.stop()
PYTHON
Der vorbereitete Befehl bigdata-python wählt die Python-Umgebung mit installiertem PySpark. Führen Sie die Datei als normales Python-Programm aus und speichern Sie ihre normale Ausgabe lokal:
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/first-run.txt
cat /home/labex/project/output/first-run.txt
Erwarten Sie eine Tabelle mit den Spalten order_id, product und quantity sowie s1/notebook/2, s2/pen/1 und s3/folder/3. Startwarnungen oder Fortschrittsmeldungen können weiterhin im Terminal erscheinen; sie sind von der gespeicherten Tabelle getrennt. Ohne ausdrückliche Sortierung ist keine Zeilenreihenfolge zugesichert. Identifizieren Sie Datensätze deshalb über ihre Bestell-IDs.
Ihr Python-Programm ist der Driver: Es erstellt die Sitzung und fordert Spark zur Arbeit auf. In einem Cluster führen Executor-Prozesse Tasks auf Datenteilen aus. Hier findet die Ausführung lokal auf demselben Rechner statt. Sie haben ein funktionierendes Programm, ohne einen Cluster konfigurieren zu müssen.
Prüfen Sie das Schema und zählen Sie Zeilen
In diesem Schritt unterscheiden Sie, was eine Spalte enthalten kann, von der Anzahl der Datensätze in der Tabelle. Ein Schema listet Spaltennamen, Datentypen und die Zulässigkeit von NULL auf. Eine Ganzzahlspalte stellt ganze Zahlen für Mengenberechnungen dar; Text hat andere Operationen. Nullable bedeutet, dass ein Feld einen fehlenden Wert zulässt, nicht dass er in jeder Zeile fehlt.
Ergänzen Sie printSchema zum Prüfen dieser Beschreibung und count zum Berechnen der Zeilenanzahl. Hier ist die vollständige aktualisierte Datei, damit kein versteckter interaktiver Zustand vorausgesetzt wird:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/schema-and-count.txt
cat /home/labex/project/output/schema-and-count.txt
Die Tabelle sollte weiterhin drei Datensätze enthalten. Suchen Sie darunter nach:
root
|-- order_id: string (nullable = true)
|-- product: string (nullable = true)
|-- quantity: integer (nullable = true)
Row count: 3
printSchema beschreibt den DataFrame. show und count sind Aktionen, die Spark auffordern, ein sichtbares Ergebnis oder einen Wert zu erzeugen. Spark kann viel Verarbeitung aufschieben, bis eine Aktion eine Antwort verlangt; spätere Labs erklären diese verzögerte Ausführung anhand von Transformationen und Plänen. Eine Anzahl von 3 bedeutet drei Bestelldatensätze, nicht die Summe der Mengen, die eine andere Geschäftsfrage beantworten würde.
Jeder Aufruf startet einen neuen Python-Prozess und erstellt seine eigene SparkSession. Die Datei erstellt df ausdrücklich; sie setzt nicht voraus, dass eine interaktive Shell zuvor spark oder df erstellt hat.
Sagen Sie einen zusätzlichen Datensatz voraus und testen Sie ihn
In diesem Schritt ändern Sie die Daten und behalten ihre Struktur bei. Sagen Sie voraus, was das Hinzufügen von s4/notebook/1 verändert: Die Anzahl der Bestellzeilen steigt auf vier, das Schema bleibt gleich, und die ursprünglichen drei Datensätze bleiben erhalten. Die neue Zeile ist eine Bestellung, obwohl notebook bereits in einer anderen Bestellung vorkommt.
Ersetzen Sie das Skript durch diese Version, die nur diesen Datensatz ergänzt:
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
("s4", "notebook", 1),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
Führen Sie die Datei erneut in einem frischen Prozess aus:
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/expanded-run.txt
cat /home/labex/project/output/expanded-run.txt
Finden Sie alle vier Bestell-IDs und Row count: 4. Das Schema sollte weiterhin dieselben drei Spaltennamen und Typen beschreiben. Die Wiederholung eines Produktnamens gruppiert Bestellungen nicht automatisch und entfernt keine Duplikate; solche Operationen müssen in späteren Lektionen ausdrücklich angefordert werden.
Die früheren Dateien first-run.txt und schema-and-count.txt bleiben als Belege Ihrer Version mit drei Zeilen erhalten. Die ausführbare Datei beschreibt nun die Version mit vier Zeilen, sodass ein weiterer unabhängiger Lauf dieses Ergebnis reproduzieren kann. Sie haben einen kleinen Zyklus aus Ändern, Vorhersagen, Ausführen und Prüfen abgeschlossen. Damit können Sie größere Datenworkflows beurteilen, statt nur auf ein erfolgreiches Programmende zu vertrauen.
Zusammenfassung
Sie haben ein echtes PySpark-Skript mit eigener SparkSession ausgeführt, einen typisierten DataFrame erstellt und seine angezeigten Datensätze, sein Schema und seine Zeilenanzahl geprüft. Ein zusätzlicher Datensatz änderte die Zeilenanzahl bei unverändertem Schema. Als Nächstes wechseln Sie von innerhalb von Python geschriebenen Datensätzen zu einer CSV-Datei mit ausdrücklichem Schema.
