Einführung
Sie haben Spark-Transformationen geschrieben, Aktionen ausgelöst und Daten gespeichert. Nun verbinden Sie diese bekannten Operationen mit Ausführungsbelegen: einem Plan, der die Arbeit beschreibt, und einem echten Ereignisprotokoll, das Jobs, Stages und Tasks aufzeichnet. Ziel ist, Verhalten zu erklären und nützliche Fehler zu lesen, statt interne Taskanzahlen auswendig zu lernen.
Der Driver führt Ihre Anwendung aus und koordiniert Arbeit. In einem Cluster führen Executors Tasks auf zugewiesenen Ressourcen aus; local[2] führt Spark-Arbeit auf diesem einen Rechner mit zwei Ausführungsthreads aus. Ein Job kann in Stages aufgeteilt werden, und jede Stage führt Tasks auf Berechnungspartitionen aus. Eine Aktion kann insbesondere durch Abfrageplanung und adaptives Verhalten mehr als einen Job verursachen; genaue Anzahlen können variieren.
Dieses unabhängige Lab stellt saubere Bestellungen, Kunden, zwei absichtlich fehlerhafte Ausdrücke und ein nur lesendes Werkzeug zur Ereigniszusammenfassung bereit. Filterung, Gruppierung, Joins und Parquet-Ausgaben sollten Sie kennen. Sie stimmen keine Ressourcen ab, sammeln nicht den allgemeinen Datensatz für Python-Schleifen ein und stellen keinen Cluster bereit. Verwenden Sie das Terminal.
Verbinden Sie eine bekannte Abfrage mit ihrem Ausführungsplan
In diesem Schritt untersuchen Sie eine tägliche regionale Abfrage bezahlter Verkäufe, bevor Sie ihre Ergebnisse anfordern. Reader, Filter, Betrag, Kunden-Join und Aggregation wiederholen frühere Kenntnisse.
Hier verwendet groupBy("order_date", "region") zwei Schlüssel gemeinsam: Jede unterschiedliche Kombination aus Datum und Region bildet eine Gruppe. East am 14. Oktober ist von East am 15. Oktober und West am 14. Oktober getrennt. Anzahlen und Summen werden innerhalb jeder Kombination berechnet. Dies verbindet die bisherigen Tages- und Regionsfragen zu einem Bericht. explain(mode="formatted") gibt die geplanten Operatoren aus, statt Berichtszeilen zurückzugeben.
Die SparkSession aktiviert außerdem ein echtes Ereignisprotokoll in output/spark-events. Komprimierung ist deaktiviert, damit dieses Übungsprotokoll nach dem Anwendungsende lesbar ist. Diese Einstellungen zeichnen die Ausführung auf; sie ersetzen sie nicht durch erfundene Jobeinträge. Das Zielverzeichnis ist vorbereitet.
Starten Sie das vollständige Untersuchungsskript:
cd /home/labex/project
cat > /home/labex/project/inspect_execution.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, sum as spark_sum, spark_partition_id
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = (
SparkSession.builder.appName("InspectRetailExecution").master("local[2]")
.config("spark.eventLog.enabled", "true")
.config("spark.eventLog.dir", "file:///home/labex/project/output/spark-events")
.config("spark.eventLog.compress", "false")
.config("spark.eventLog.rolling.enabled", "false")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
matched = paid.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(
col("o.order_date").alias("order_date"), col("c.region").alias("region"), col("o.order_amount").alias("order_amount")
)
report = matched.groupBy("order_date", "region").agg(
count("*").alias("paid_order_count"), spark_sum("order_amount").alias("revenue")
)
print("Report execution plan:")
report.explain(mode="formatted")
print("Report action result:")
report.orderBy("order_date", "region").show(truncate=False)
report.write.mode("overwrite").parquet("/home/labex/project/output/daily-regional-sales")
print("Computation partition IDs represented by source rows:")
orders.select(spark_partition_id().alias("compute_partition")).dropDuplicates().orderBy("compute_partition").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_execution.py > /home/labex/project/output/execution-plan-and-result.txt
cat /home/labex/project/output/execution-plan-and-result.txt
Suchen Sie unter Report execution plan nach den Rollen von Filterung, Projektion, Join, Aggregation und eventuellen Exchange-Operatoren, die Datensätze zwischen Berechnungspartitionen bewegen. Namen und Planstruktur können je nach Optimierung variieren. Ein Exchange ist eine Ausführungsgrenze, kein neues Ausgabeverzeichnis.
Unter Report action result erwarten Sie 14. Oktober / East / 1 / 20.00, 14. Oktober / West / 1 / 25.00, 15. Oktober / East / 1 / 50.00 und 15. Oktober / West / 1 / 30.00. Die Summe bleibt 125.00. show und der Writer fordern die tatsächliche Ausführung der beschriebenen Arbeit an; auch Planung und Behandlung von Eingabemetadaten können vorbereitende Arbeit leisten.
Die letzte Anzeige verwendet spark_partition_id, um IDs der Berechnungspartitionen zu zeigen, die durch diese Quellzeilen vertreten sind. dropDuplicates listet jede beobachtete ID einmal auf. IDs und ihre beobachtete Anzahl können variieren; leere Partitionen müssen nicht erscheinen. Diese Laufzeitgruppen unterscheiden sich von den mit partitionBy geschriebenen Datumsverzeichnissen des vorherigen Labs. Das eine ist Arbeitsorganisation, das andere die gespeicherte Datenstruktur.
Lesen Sie echte Job-, Stage- und Task-Ereignisse
In diesem Schritt prüfen Sie das Ereignisprotokoll der gerade ausgeführten Anwendung. Das bereitgestellte Werkzeug liest JSON-Ereignisdatensätze aus diesem Verzeichnis und fasst tatsächliche Ereignisse zusammen. Sie müssen keinen Protokollparser schreiben oder einen versionsspezifischen Dateinamen auswählen.
Führen Sie das Zusammenfassungswerkzeug aus und übergeben Sie das Ereignisverzeichnis als Eingabeargument. Das Shell-Zeichen > erhält seine Ausgabe zur Prüfung:
bigdata-python /home/labex/project/tools/summarize-events.py /home/labex/project/output/spark-events > /home/labex/project/output/event-summary.txt
cat /home/labex/project/output/event-summary.txt
Erwarten Sie eine tatsächliche Anwendungskennung, positive Anzahlen für Jobs started, Stages submitted und Tasks ended sowie Job-zu-Stage-Verweise. Die Zahlen hängen von diesem Lauf ab und können sich durch Optimierung, Dateiaufteilung und Spark-Version ändern. Sie sind beobachtete Belege, keine exakt zu reproduzierenden Zielwerte.
Eine Aktion fordert eine Antwort an; Spark kann die nötige Arbeit in mehrere Jobs organisieren. Ein Job verweist auf Stages; Stages enthalten Tasks für Arbeitspartitionen. Shuffle-Grenzen können Stages trennen, während mehrere Transformationen innerhalb einer Stage zusammengefasst werden können. local[2] begrenzt gleichzeitige Ausführungsthreads; es bedeutet nicht, dass jede Anwendung genau zwei Jobs, Stages oder Tasks hat.
Bei winzigen Daten können JVM-Start, Ablaufplanung und Dateibehandlung gegenüber der eigentlichen Berechnung deutlich auffallen. Ein kleiner Laufzeitunterschied ist keine verlässliche Tuning-Schlussfolgerung. Verwenden Sie zuerst Plan- und Ereignisbelege, um die Arbeit zu verstehen.
Diagnostizieren und korrigieren Sie eine unbekannte Spalte
In diesem Schritt lesen Sie einen absichtlichen Namensauflösungsfehler, bevor Sie dessen Ursache korrigieren. Prüfen Sie das vorbereitete fehlerhafte Beispiel:
cat /home/labex/project/tools/broken-column.py
Das Schema deklariert quantity, während das abschließende select quanity anfordert. Speichern Sie die tatsächliche Fehlerausgabe. > leitet normale Ausgabe in die Protokolldatei; 2>&1 sendet Standardfehler an dasselbe Ziel, sodass auch der Traceback gespeichert wird. Der nächste Befehl soll fehlschlagen.
bigdata-python /home/labex/project/tools/broken-column.py > /home/labex/project/output/column-error.log 2>&1
tail -n 20 /home/labex/project/output/column-error.log
Finden Sie den nicht auflösbaren Spaltennamen und mögliche vorgeschlagene gültige Namen. Ein langer Traceback enthält Implementierungsdetails, aber der Spaltenverweis zeigt die Korrektur. Die Eingabe benötigt weder Neuinstallation noch eine Umwandlung in eine Python-Schleife.
Erstellen Sie ein korrigiertes Programm mit dem deklarierten Spaltennamen und behalten Sie das fehlerhafte Übungsbeispiel zum Vergleich:
cat > /home/labex/project/fixed_column.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("OrderExpressionExample").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
fixed = orders.select("order_id", "quantity")
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-quantities")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_column.py > /home/labex/project/output/fixed-column-result.txt
cat /home/labex/project/output/fixed-column-result.txt
Erwarten Sie alle sechs IDs mit Mengen 2, 1, 1, 2, 1 und 2. Die Projektion lässt sich nun anhand des tatsächlichen Schemas auflösen und speichert ihr echtes Ergebnis. Spaltennamen einschließlich Schreibweise sind Teil der Datenvereinbarung.
Korrigieren Sie einen Filter ohne boolesche Bedingung
In diesem Schritt unterscheiden Sie eine gültige Ganzzahlspalte von einer gültigen Filterbedingung. Prüfen Sie das zweite vorbereitete Beispiel:
cat /home/labex/project/tools/broken-filter.py
filter erwartet einen booleschen Ausdruck pro Zeile. col("quantity") allein ist eine Ganzzahl und kein Wahr-/Falsch-Test. Erfassen Sie den absichtlichen Typkonflikt. Der nächste Befehl soll fehlschlagen.
bigdata-python /home/labex/project/tools/broken-filter.py > /home/labex/project/output/filter-error.log 2>&1
tail -n 20 /home/labex/project/output/filter-error.log
Suchen Sie die Filterbedingung und ihren nichtbooleschen Ganzzahltyp. Die Korrektur besteht darin, den vorgesehenen Vergleich quantity > 0 mit dem bereits gelernten Spaltenoperator auszudrücken. Führen Sie das korrigierte vollständige Programm aus:
cat > /home/labex/project/fixed_filter.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("OrderExpressionExample").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
fixed = orders.filter(col("quantity") > 0)
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-orders")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_filter.py > /home/labex/project/output/fixed-filter-result.txt
cat /home/labex/project/output/fixed-filter-result.txt
Alle sechs sauberen Beispielzeilen haben positive Mengen und bleiben daher erhalten. Eine negative Menge oder null würde diese Bedingung nicht erfüllen; diese konkrete Korrektur wendet nicht die separate Regel für bezahlte Verkäufe an.
Sie haben nun Belege auf drei Ebenen: vorgesehener Ausdruck und Schema, tatsächlich geplante beziehungsweise ausgeführte Arbeit und korrekte gespeicherte Ergebnisse nach gezielten Korrekturen. Behalten Sie bei der Erklärung der Ausführung die Unterscheidung zwischen gespeicherten Datumsverzeichnissen und Berechnungspartitionen bei. Als Nächstes übermitteln Sie ein unabhängiges Batch-Skript mit ausdrücklichen Ein- und Ausgabeparametern.
Zusammenfassung
Sie haben bekannte Transformationen mit echtem Plan und Ereignisprotokoll verbunden, Anwendungs-, Job-, Stage- und Task-Belege ohne festgelegte Anzahlen beobachtet und Berechnungspartitionen von Datumsverzeichnissen unterschieden. Eine nicht auflösbare Spalte und einen nichtbooleschen Filter haben Sie anhand von Schema- und Fehlerbelegen korrigiert. Als Nächstes machen ausdrückliche Übermittlungsparameter Ihre Batch-Anwendung unabhängig von einem vorherigen Prozess.
