Einführung
Eine nützliche Batch-Anwendung sollte aus einem gespeicherten Programm mit ausdrücklichen Ein- und Ausgaben laufen, statt von einem DataFrame oder einer temporären Ansicht einer früheren Sitzung abzuhängen. Sie übermitteln ein vorbereitetes Tagesverkaufsprogramm in getrennten Prozessen und prüfen seine gespeicherten Ergebnisse und Ausführungsbelege.
spark-submit ist Sparks Anwendungsstarter. Sie geben Ausführungs-Master, Python-Programm und dessen Ein-/Ausgabeargumente an. Das Skript lässt bewusst diese externe Master-Einstellung die Ausführung steuern. Seine Berichtslogik wiederholt typisiertes Lesen, paid-Filter, Dezimalbetrag und Gruppierung aus früheren Labs; die Argumentverarbeitung ist bereitgestellt, damit Sie sich auf Übermittlung und sichere Wiederholung konzentrieren können.
Dieses unabhängige Lab liefert zwei gültige CSV-Lieferungen und das vollständige Programm. Sie prüfen Protokolle und Rückgabecodes, ändern den Eingabepfad und unterscheiden fehlende Eingabe von einem Konflikt mit vorhandener Ausgabe. Remote-Cluster, Paketierung, Scheduler, Hive oder Zustand eines früheren Prozesses sind nicht erforderlich. Verwenden Sie das Terminal.
Übermitteln Sie ein vorbereitetes Programm mit ausdrücklichen Argumenten
In diesem Schritt verarbeiten Sie die erste Lieferung mit einer eigenständigen Anwendung. Prüfen Sie Quelle und bereitgestelltes Programm:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
cat /home/labex/project/sales_batch.py
Der Parser akzeptiert bereits input_path, output_path und optional --overwrite. Sie müssen argparse nicht implementieren. Der erste Positionspfad ist immer die Eingabe, der zweite die Ausgabe. Der SparkSession-Builder ruft master nicht auf, sodass die Einstellung --master des Starters die Ausführung steuert.
Lesen Sie den Befehl von links nach rechts: spark-submit-Optionen stehen vor dem Python-Dateinamen; Argumente danach gehen an die Anwendung. Setzen Sie local[2] in Anführungszeichen, damit die Shell seine Klammern wörtlich behandelt. > speichert normale Ausgabe, und 2>&1 sendet Fehlerausgabe an dasselbe Protokoll:
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/orders.csv /home/labex/project/output/run-1 > /home/labex/project/output/run-1-job.log 2>&1
Erfassen Sie sofort den Rückgabestatus, bevor Sie einen anderen Terminalbefehl ausführen. $? ist der Status des vorherigen Befehls: null bedeutet Erfolg, ungleich null bedeutet Fehler. printf schreibt diesen Wert und einen Zeilenumbruch in eine lokale Statusdatei:
printf '%s\n' "$?" > /home/labex/project/output/run-1-exit-status.txt
cat /home/labex/project/output/run-1-exit-status.txt
Erwarten Sie 0. Lesen Sie das Ende des Driver-Protokolls:
tail -n 30 /home/labex/project/output/run-1-job.log
Finden Sie Master: local[2], die tatsächliche Anwendungs-ID, die gewählten Ein-/Ausgabepfade und die Berichtsdaten. Die ID variiert. Der Bericht sollte 14. Oktober / 2 / 45.00 und 15. Oktober / 2 / 80.00 zeigen. Ein Rückgabecode null zusammen mit einem korrekten tatsächlichen Parquet-Ergebnis ist ein stärkerer Beleg als eine beruhigende Protokollzeile.
Das Programm erstellt und beendet seine eigene Sitzung. Es verwendet keine Ansichten oder Python-Variablen der vorherigen Lektion wieder.
Ändern Sie die Eingabe in einem neuen Prozess
In diesem Schritt weisen Sie nach, dass das Programm sein Ergebnis aus dem übergebenen Pfad ableitet. Prüfen Sie die zweite Lieferung:
cat /home/labex/project/data/new-orders.csv
Sie hat zwei bezahlte Bestellungen am 16. Oktober mit Beträgen 30.00 und 25.00 sowie eine ausstehende Bestellung. Sagen Sie eine Berichtszeile voraus: Anzahl 2 und Umsatz 55.00. Behalten Sie die erste Ausgabe und übermitteln Sie diese neue Datei an ein separates Ziel:
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/new-orders.csv /home/labex/project/output/run-2 > /home/labex/project/output/run-2-job.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/run-2-exit-status.txt
cat /home/labex/project/output/run-2-exit-status.txt
tail -n 30 /home/labex/project/output/run-2-job.log
Erwarten Sie Status 0 und das vorhergesagte Ergebnis vom 16. Oktober. Die tatsächliche Anwendungs-ID unterscheidet sich vom ersten Prozess. Das Programm lud den neuen Pfad und erstellte seine eigene Sitzung; eine zurückgebliebene Ansicht war nicht erforderlich.
run-1 und run-2 bleiben beide verfügbar, sodass die Beziehung jeder Eingabe zu ihrem Ergebnis prüfbar ist. Ändern Sie die Quelldatei nicht bloß, um eine gemerkte Antwort erscheinen zu lassen.
Diagnostizieren Sie fehlende Eingabe und vorhandene Ausgabe getrennt
In diesem Schritt erfassen Sie zwei unterschiedliche Pfadfehler, ohne Daten zu löschen. Übermitteln Sie zuerst eine nicht vorhandene Eingabe. Der nächste Befehl soll fehlschlagen.
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/missing-orders.csv /home/labex/project/output/unused-missing-input > /home/labex/project/output/missing-input.log 2>&1
Erfassen Sie sofort ihren Status vor der Protokollprüfung:
printf '%s\n' "$?" > /home/labex/project/output/missing-input-exit-status.txt
cat /home/labex/project/output/missing-input-exit-status.txt
tail -n 25 /home/labex/project/output/missing-input.log
Erwarten Sie einen Status ungleich null und eine Diagnose zum nicht gefundenen Pfad mit dem Namen missing-orders.csv. Den Eingabepfad zu korrigieren löst dieses Problem; Master-Änderung oder Neuerstellen eines Clusters tun es nicht.
Der Standardmodus errorifexists des Skripts schützt auch ein vorhandenes Ziel. Versuchen Sie die bekannte Eingabe mit dem bereits abgeschlossenen Pfad run-1. Der nächste Befehl soll fehlschlagen.
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/orders.csv /home/labex/project/output/run-1 > /home/labex/project/output/existing-output.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/existing-output-exit-status.txt
cat /home/labex/project/output/existing-output-exit-status.txt
tail -n 25 /home/labex/project/output/existing-output.log
Erwarten Sie einen Status ungleich null und eine Diagnose zur bereits vorhandenen Ausgabe. Dies unterscheidet sich von einer fehlenden Eingabe, und der frühere erfolgreiche Bericht sollte unverändert bleiben.
Nehmen Sie eine Korrektur vor, die vorhandene Ergebnisse erhält: mit korrekter Quelle und neuem Ausgabeverzeichnis:
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/orders.csv /home/labex/project/output/run-3 > /home/labex/project/output/run-3-job.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/run-3-exit-status.txt
cat /home/labex/project/output/run-3-exit-status.txt
tail -n 25 /home/labex/project/output/run-3-job.log
Erwarten Sie Erfolg und die ursprünglichen beiden Tagessummen. Ein neues Ziel erhält beide früheren Lieferungen und die Fehlerbelege, ohne Rohdateien oder große Ausgabebäume zu entfernen.
Erstellen Sie einen vorgesehenen abgeleiteten Datenstand ausdrücklich neu
In diesem Schritt wählen Sie bewusst das Ersetzen eines bekannten vollständigen Datenstands. Das optionale Argument --overwrite ist bereits implementiert: Es ändert nur den Zielschreibmodus dieses Programms zu overwrite. Es verändert weder Eingabe noch andere Ergebnispfade oder Sparks externen Master.
Verwenden Sie es, wenn Sie das gesamte angegebene abgeleitete Ergebnis aus derselben Quelle neu erstellen möchten. Hier ist run-1 dieser bekannte Datenstand; die anderen Läufe und Rohdateneingaben bleiben getrennt:
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/orders.csv /home/labex/project/output/run-1 --overwrite > /home/labex/project/output/overwrite-job.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/overwrite-exit-status.txt
cat /home/labex/project/output/overwrite-exit-status.txt
tail -n 30 /home/labex/project/output/overwrite-job.log
Erwarten Sie Status 0, Write mode: overwrite und dieselben beiden Berichtszeilen. Dieser neue Prozess ersetzt den Datenstand, statt doppelte Tage hinzuzufügen. Seine Anwendungs-ID unterscheidet sich vom ersten Lauf, und sein Protokoll hat einen eigenen Dateinamen.
Sie haben zwei gültige Reaktionen auf vorhandene Ausgabe geübt: einen neuen Pfad wählen, wenn beide Versionen erhalten bleiben sollen, oder einen bekannten abgeleiteten Datenstand ausdrücklich neu erstellen, wenn Ersetzen beabsichtigt ist. Eingabepfadfehler benötigen weiterhin eine korrekte Eingabe; overwrite repariert keine fehlenden Daten. Diese Vorgehensweisen für Übermittlung und Wiederholung bereiten Sie auf die abschließende Challenge einer wiederverwendbaren Verkaufspipeline vor.
Zusammenfassung
Sie haben eine parametrisierte Spark-Anwendung in unabhängigen Prozessen gestartet, tatsächlichen Status und Protokolle erfasst und Ergebnisse verschiedener Eingaben geprüft. Sie haben fehlende Eingaben von vorhandenen Ausgabekonflikten unterschieden und entweder ein neues Ziel oder ausdrückliches Ersetzen eines Datenstands gewählt. Die letzte Challenge verbindet diese Übermittlungsgewohnheiten mit Bereinigung, Anreicherung, Aggregation und partitionierter Ausgabe.
