Einführung
Eine korrekt typisierte Bestelltabelle ist ein Ausgangspunkt, aber noch keine Antwort. Ein Kollege möchte vielleicht nur wenige Spalten, ausschließlich bezahlte Bestellungen und einen aus Menge und Einzelpreis berechneten Betrag. Sie drücken diese Entscheidungen als Spark-DataFrame-Operationen aus, während die ursprünglichen Datensätze verfügbar bleiben.
Ein Spaltenausdruck beschreibt eine Berechnung oder Bedingung, die Spark auf Zeilen anwenden soll. Sie wählen Felder aus, kombinieren Bedingungen, ergänzen einen Dezimalbetrag und geben einer Ergebnisspalte einen lesbaren Namen. Diese Operationen erzeugen neue DataFrames; Aktionen wie show fordern tatsächliche Ergebnisse an. Die Unterscheidung hilft, Sparks verzögerte Ausführung zu erklären, ohne bereits interne Ablaufplanung zu benötigen.
Dieses Lab stellt unabhängig sechs saubere Handelsbestellungen bereit. Das ausdrückliche CSV-Schema aus dem vorherigen Lab und grundlegendes Python sollten Sie verstehen. Jede ausführbare Version enthält den vollständigen Reader und die SparkSession, daher ist kein früherer Prozess oder eine frühere Datei nötig. Verwenden Sie das Terminal und die vorbereitete Umgebung bigdata-python.
Wählen Sie die für eine Frage benötigten Spalten aus
In diesem Schritt grenzen Sie die Tabellenspalten ein und behalten alle sechs Bestelldatensätze. Beginnen Sie mit der Quelle:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
select wählt, welche Spalten ein neuer DataFrame in welcher Reihenfolge bereitstellt. Es filtert keine Zeilen. Das vollständige Skript wiederholt den bekannten ausdrücklichen Reader und wählt dann Bestellkennung, Menge, Preis und Status aus. Datums- und Kundenfelder bleiben in df, obwohl selected sie nicht bereitstellt.
Speichern Sie das Skript mit einem Here-Dokument in Anführungszeichen und führen Sie es aus. > schreibt die Datei oder ersetzt die vorherige Version; der abschließende Marker PYTHON beendet den Quelltext:
cat > /home/labex/project/transform_orders.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("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
selected = df.select("order_id", "quantity", "unit_price", "order_status")
selected.orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/selected-orders.txt
cat /home/labex/project/output/selected-orders.txt
Erwarten Sie alle sechs Bestell-IDs mit vier angezeigten Spalten. orderBy macht die Anzeige vorhersehbar, während show die zum Zurückgeben der Zeilen erforderliche Arbeit auslöst. Dies ist eine Spaltenauswahl, auch Projektion genannt. Keine Quelldatei wurde verändert und keine Bestellung entfernt.
Kombinieren Sie Bedingungen zur Zeilenauswahl
In diesem Schritt behalten Sie bezahlte Bestellungen mit einer Menge von mindestens zwei. col("order_status") verweist auf eine Spark-Spalte; der Vergleich mit == "paid" baut eine Bedingung für Spark auf statt eines einzelnen Python-Werts True oder False. >= erstellt die Mengenbedingung.
Verwenden Sie &, um Spark-Spaltenbedingungen mit UND-Bedeutung zu verbinden. Setzen Sie jeden Vergleich in Klammern, weil die Operatorrangfolge von Python wichtig ist. Das gewöhnliche Python-and ist nicht der Operator zum Kombinieren von Spark-Column-Objekten.
Sagen Sie die passenden IDs voraus: o1, o4 und o6 erfüllen beide Bedingungen. o2 ist bezahlt, hat aber Menge eins. Ersetzen und starten Sie das vollständige Skript:
cat > /home/labex/project/transform_orders.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("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
eligible = df.filter((col("order_status") == "paid") & (col("quantity") >= 2))
eligible.select("order_id", "quantity", "unit_price", "order_status").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/eligible-orders.txt
cat /home/labex/project/output/eligible-orders.txt
Das Ergebnis sollte nur diese drei IDs enthalten, jeweils mit Menge 2. filter ändert die ausgewählten Zeilen; select steuert die angezeigten Spalten. Sie können beide Operationen kombinieren, während df weiterhin die vollständige Quelle mit sechs Zeilen beschreibt.
Die Mengenschwelle dieses Schritts ist eine Demonstrationsbedingung und keine allgemeine Verkaufsdefinition. Die nächste Berechnung berücksichtigt alle vier bezahlten Bestellungen einschließlich o2, weil jede bezahlte Bestellung Umsatz beiträgt.
Berechnen und benennen Sie dezimale Bestellbeträge
In diesem Schritt berechnen Sie den Betrag jeder bezahlten Bestellung als Menge mal dezimalem Einzelpreis. withColumn fügt einem neuen DataFrame eine benannte Spalte hinzu oder ersetzt sie, wenn ihr Name bereits existiert. Hier wird order_amount mit Sparks eingebautem Multiplikationsausdruck berechnet.
Das folgende select behält die Bestell-ID und gibt der berechneten Spalte mit alias den Anzeigenamen amount. Ein Alias ändert den Spaltennamen des Ergebnisses, nicht die CSV-Kopfzeile. Alle vier bezahlten Bestellungen gehören in diese Berechnung; pending und refunded werden ausgeschlossen, ohne Erstattungen in dieser vereinfachten Regel als negativen Umsatz zu behandeln.
Führen Sie diese vollständige Version aus:
cat > /home/labex/project/transform_orders.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("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid_result = (
df.filter(col("order_status") == "paid")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/paid-amounts.txt
cat /home/labex/project/output/paid-amounts.txt
Erwarten Sie o1/20.00, o2/25.00, o4/50.00 und o6/30.00. Zusätzliche angehängte Nullen können erscheinen. Das Schema sollte amount als Dezimaltyp kennzeichnen. Spark kann bei der Multiplikation die Dezimalpräzision erweitern; entscheidend ist, dass die Berechnung Dezimalarithmetik behält, statt Preise in binäre Gleitkommazahlen umzuwandeln.
filter, withColumn, select und orderBy beschreiben Transformationen und liefern neue DataFrames. Die Definition dieser Kette materialisiert nicht sofort jede Zwischentabelle. show ist eine Aktion, die Spark auffordert, die erforderliche Arbeit auszuführen und eine Antwort anzuzeigen. printSchema beschreibt die Spaltenstruktur. Später untersuchen Sie den tatsächlichen Ausführungsplan hinter einer solchen Kette.
Ändern Sie einen Status und sagen Sie die Wirkung voraus
In diesem Schritt behalten Sie das paid-Ergebnis und ergänzen eine separate Übungsansicht für ausstehende Bestellungen. Sagen Sie die Ausgabe vor der Ausführung voraus: Nur o3 ist pending, mit Menge eins und Preis 10.00, also beträgt sein Betrag 10.00.
Das neue pending_result verwendet dieselbe Ausdruckskette mit nur geändertem Statuswert. Es beginnt bei df, nicht beim bereits gefilterten paid_result, sodass ausstehende Datensätze noch verfügbar sind:
cat > /home/labex/project/transform_orders.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("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid_result = (
df.filter(col("order_status") == "paid")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()
pending_result = (
df.filter(col("order_status") == "pending")
.withColumn("order_amount", col("quantity") * col("unit_price"))
.select("order_id", col("order_amount").alias("amount"))
)
print("Pending practice:")
pending_result.orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/status-comparison.txt
cat /home/labex/project/output/status-comparison.txt
Unter Paid orders sollten dieselben vier bezahlten Beträge bleiben. Unter Pending practice erwarten Sie nur o3/10.00. Würden Sie die paid-Ansicht nach pending filtern, erhielten Sie keine Zeilen, weil diese Datensätze bereits ausgeschlossen wurden; der passende Ausgangspunkt zählt.
Beide Ketten werden durch Sparks eingebaute Ausdrücke ausgewertet. Sie haben den allgemeinen Datensatz nicht in einer Python-Liste gesammelt und in einer Schleife verarbeitet. So bleibt die Verarbeitungsbeschreibung für Sparks Ausführungsengine geeignet. Ihr abschließendes Skript behält die paid-Berechnung und den kleinen Übungsvergleich, während die ursprüngliche CSV weiterhin alle sechs Bestellungen enthält.
Zusammenfassung
Sie haben Spalten ausgewählt, Zeilenbedingungen mit Spark-Ausdrücken kombiniert, dezimale Bestellbeträge berechnet und Ausgabespalten benannt. Der Vergleich von paid- und pending-Ansichten zeigte, warum Transformationen von der passenden Quelle ausgehen sollten und wie Aktionen Ergebnisse anfordern. Als Nächstes wenden Sie bekannte Ideen zum Lesen und Auswählen auf JSON-Eingaben an.
