Einführung
Der Einzelhändler hat bereits einen korrekten, typisierten Satz bezahlter Bestellergebnisse. Nun muss eine andere Anwendung diese Ergebnisse später lesen, ein Datum auswählen und darauf vertrauen können, dass ein wiederholter Export Bestellungen nicht unbemerkt dupliziert hat. Daten zu speichern ist Teil der Verarbeitungsvereinbarung und nicht bloß die letzte Skriptzeile.
Sie schreiben Parquet mit Verzeichnispartitionierung nach order_date. Das Wurzelverzeichnis des Datensatzes enthält nach Datum benannte Unterverzeichnisse, und Spark rekonstruiert die Datumsspalte beim Lesen dieser Wurzel. Diese Speicherverzeichnisse unterscheiden sich von den Berechnungspartitionen zur Arbeitsausführung; das nächste Lab untersucht die Ausführung separat.
Dieses Lab bereitet unabhängig vier bezahlte Bestellungen mit Dezimalbeträgen als lokale Parquet-Eingabe vor. DataFrame-Lesen, Filter und den minimalen Überschreib-Parquet-Ablauf sollten Sie kennen. Hive-Katalog, HDFS, SQL-Vorkenntnisse, Installation oder frühere Ausgaben sind nicht erforderlich. Verwenden Sie das Terminal.
Schreiben Sie einen datumspartitionierten Datensatz und lesen Sie ihn zurück
In diesem Schritt ordnen Sie die vorbereiteten Ergebnisse in Datumsverzeichnisse ein. Der Eingabedatensatz unter data/prepared-orders enthält vier Bestellungen mit derselben typisierten Menge, Dezimalpreis, Datum und dezimalem order_amount wie zuvor. Parquet ist binär; prüfen Sie es mit Spark, statt von cat eine CSV-Textanzeige zu erwarten.
Beginnen Sie in Ihrem Arbeitsverzeichnis:
cd /home/labex/project
partitionBy("order_date") weist den Writer an, Zeilen in nach Datum benannte Verzeichnisse zu legen. mode("overwrite") ersetzt den vorhandenen abgeleiteten Datensatz am angegebenen Ausgabepfad. Halten Sie diesen Ausgabepfad von input_path getrennt, das unverändert bleiben muss.
Speichern und starten Sie das vollständige Programm:
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/partitioned-read-back.txt
cat /home/labex/project/output/partitioned-read-back.txt
Erwarten Sie vier rekonstruierte Datensätze: o1 und o2 am 14. Oktober sowie o4 und o6 am 15. Oktober mit Beträgen 20.00, 25.00, 50.00 und 30.00. Main rows sollte 4 sein. Prüfen Sie quantity als integer, Preis und Betrag als decimal und order_date als date.
Eine Partitionsspalte kann nach dem Lesen an einer anderen Position stehen, weil Spark sie aus Verzeichnisnamen rekonstruiert, statt sie als gewöhnliches Feld in jeder Partitionsdatei zu speichern. Spaltennamen und Werte bestimmen ihre Identität; verwenden Sie ein ausdrückliches select, wenn ein nachfolgender Verbraucher eine bestimmte Spaltenreihenfolge verlangt. Auch die Anzeigereihenfolge benötigt orderBy.
Prüfen Sie Verzeichnisse und wählen Sie ein Datum aus
In diesem Schritt verknüpfen Sie die Verzeichnisstruktur mit der rekonstruierten Datumsspalte. Listen Sie die Ausgabe rekursiv auf; -R schließt die Datumsunterverzeichnisse und ihre Dateien ein:
ls -R /home/labex/project/output/orders-by-date
Erwarten Sie die Verzeichnisse order_date=2026-10-14 und order_date=2026-10-15. part-Dateinamen, ihre Anzahl und versteckte Metadaten können variieren; codieren Sie keine einzelne part-Datei als Datensatz fest ein. Die maßgebliche Wurzel ist orders-by-date.
Beim Lesen dieser Wurzel kann Spark die Partitionswerte erkennen. Diese Erkennung der Dateistruktur erfordert keine Registrierung einer Hive-Partition oder -Tabelle. Das folgende unabhängige Skript liest die Wurzel, vergleicht ihre DateType-Spalte mit einer gültigen ISO-Datumszeichenfolge und speichert die ausgewählten Tagesdatensätze separat:
cat > /home/labex/project/inspect_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("InspectDateDataset").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
root = "/home/labex/project/output/orders-by-date"
df = spark.read.parquet(root)
selected = df.filter(col("order_date") == "2026-10-14")
selected.orderBy("order_id").show(truncate=False)
selected.write.mode("overwrite").parquet("/home/labex/project/output/selected-day")
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_sales_data.py > /home/labex/project/output/selected-date.txt
cat /home/labex/project/output/selected-date.txt
Erwarten Sie nur o1 und o2 mit ihren Daten vom 14. Oktober und Beträgen 20.00 und 25.00. Das andere Datum existiert weiterhin im Hauptdatensatz. Ein gefilterter Lesevorgang ändert die zurückgegebenen Datensätze; er löscht keine nicht ausgewählten Verzeichnisse.
Sie haben Speicherorganisation und tatsächliches Abfrageergebnis untersucht. Bei diesem winzigen Beispiel belegt die verstrichene Zeit nicht, dass ein Format oder Aufbau schneller ist. Ziel ist ein korrekter, auffindbarer Datensatz, den spätere Reader verwenden können.
Vergleichen Sie wiederholtes Überschreiben und Anhängen gezielt
In diesem Schritt bewahren Sie den korrekten Hauptdatensatz auf und experimentieren in einem separaten Verzeichnis. Sagen Sie die Übungsanzahlen voraus: Dieselben vier Zeilen zweimal mit overwrite zu schreiben lässt jedes Mal vier Zeilen übrig; eine anschließend angehängte vollständige Kopie erzeugt acht Zeilen. Diese zusätzlichen Zeilen sind doppelte Exporte, keine neuen Bestellungen.
Führen Sie diese vollständige Version aus. Der Hauptpfad verwendet overwrite. Nur write-mode-demo erhält das absichtliche append:
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
practice_path = "/home/labex/project/output/write-mode-demo"
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("First overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("Second overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("append").partitionBy("order_date").parquet(practice_path)
print("After append rows:", spark.read.parquet(practice_path).count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Suchen Sie Main rows: 4, First overwrite rows: 4, Second overwrite rows: 4 und After append rows: 8. Der Übungsdatensatz enthält nun absichtlich zwei Kopien jeder Bestellung und zeigt damit, warum das Anhängen eines vollständigen Datenstands für den Hauptexport ungeeignet ist.
Führen Sie das gespeicherte Programm erneut aus, um die Wiederholbarkeit des gesamten Ablaufs zu prüfen:
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Dieselben Anzahlen sollten erscheinen. Jeder vollständige Lauf erstellt den Hauptdatensatz neu und setzt den Übungsversuch vor seinem absichtlichen Anhängen zurück, sodass er nicht auf zwölf oder sechzehn Zeilen wächst. Anhängen ist eine eigene Regel für tatsächlich neue, nicht überlappende Daten; Überschreiben ist die gewählte Regel für diesen vollständigen abgeleiteten Datenstand.
Der ursprüngliche Datensatz prepared-orders bleibt verfügbar. Datumsverzeichnisse beschreiben gespeicherte Werte, während Sparks Laufzeit-Berechnungspartitionen beschreiben, wie ein Job Arbeit aufteilt. Diese zweite Bedeutung von Partition verknüpfen Sie als Nächstes mit Plänen und tatsächlicher Ausführung.
Zusammenfassung
Sie haben einen datumspartitionierten Parquet-Datensatz geschrieben, seine typisierten Datensätze aus der Wurzel rekonstruiert und ein Datum ausgewählt, ohne andere zu entfernen. Wiederholtes Überschreiben erhielt den Hauptdatenstand, während absichtliches Anhängen in der Übung doppelte Kopien demonstrierte. Als Nächstes erklären Ausführungspläne und echte Ereignisse Berechnungspartitionen, Jobs, Stages und Tasks.
