Einführung
Eine Bestellung kennt ihre Kunden- und Produktkennungen, während separate Referenztabellen den Wohnort des Kunden und die Produktkategorie beschreiben. Ein Join verbindet Datensätze über diese Schlüssel, damit ein Bericht Fragen wie Umsatz nach Region oder Produktkategorie beantworten kann.
Sie reichern bezahlte Bestellungen nacheinander mit jeder Zuordnung an, wählen eindeutige Ausgabespalten und gleichen Summen ab. Außerdem vergleichen Sie INNER und LEFT JOIN bei einem fehlenden Kunden. Die bereitgestellten Kunden- und Produktschlüssel sind eindeutig; jede zuordenbare Bestellung sollte in jeder Referenztabelle einen Datensatz finden. Doppelte Referenzschlüssel könnten verknüpfte Zeilen vervielfachen und Umsatz aufblähen, daher ist die beabsichtigte Beziehung wichtig.
Dieses unabhängige Lab bereitet saubere Bestellungen, zwei Kunden, drei Produkte und ein separates Beispiel mit fehlender Zuordnung vor. Typisiertes Lesen, Filter, Dezimalbeträge, Aggregation und minimale Parquet-Ausgabe sollten Sie kennen. Hive, SQL, Hadoop, pandas oder Clustereinrichtung sind nicht erforderlich. Verwenden Sie das Terminal.
Ergänzen Sie Kundenregionen über einen ausdrücklichen Schlüssel
In diesem Schritt verknüpfen Sie die vier bezahlten Bestellungen mit ihren eindeutigen Kundendatensätzen. Prüfen Sie die Quellbeziehung:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
cat /home/labex/project/data/customers.csv
Mehrere Bestellungen können dieselbe customer_id haben, während die Referenz je eine Zeile für c1 und c2 enthält. Gleiche Kennungen verbinden Datensätze; gleiche Zeilenpositionen wären keine gültige Regel. c1 liefert East und c2 West.
DataFrame join erwartet den anderen DataFrame, eine Zuordnungsbedingung und einen Join-Typ. inner behält nur passende Datensätze. DataFrame alias("o") und alias("c") geben den Eingaben kurze Referenznamen. col("o.customer_id") kennzeichnet die Spalte auf Bestellseite, col("c.customer_id") die auf Referenzseite.
Beide Eingaben enthalten eine customer_id-Spalte. Das ausdrückliche select wählt den Schlüssel der Bestellseite und jedes gewünschte Feld und vermeidet mehrdeutige doppelte Namen. Ein DataFrame-Alias benennt eine Eingabereferenz; ein Spaltenalias benennt ein Ergebnisfeld. Speichern und starten Sie diese vollständige Version, deren Reader und Betragsberechnung frühere Kenntnisse wiederholen:
cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("EnrichOrders").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")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")
with_customers = (
paid.alias("o")
.join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
.select(
col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
col("c.region").alias("region"),
)
)
with_customers.orderBy("order_id").show(truncate=False)
with_customers.write.mode("overwrite").parquet("/home/labex/project/output/orders-with-regions")
spark.stop()
PYTHON
bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/customer-join.txt
cat /home/labex/project/output/customer-join.txt
Erwarten Sie vier angereicherte bezahlte Zeilen. o1 und o4 sollten East sein, o2 und o6 West; ihre Beträge bleiben 20.00, 25.00, 50.00 und 30.00. Vier bezahlte Eingaben mit vollständigen eindeutigen Zuordnungen erzeugen vier verknüpfte Zeilen. Die Parquet-Ausgabe erhält die typisierten Felder und ergänzt region, ohne die Roh-CSV zu ändern.
Ergänzen Sie Produktkategorien und gleichen Sie Umsatz ab
In diesem Schritt ergänzen Sie eine zweite Referenz über product_id. Prüfen Sie die Produktzuordnung:
cat /home/labex/project/data/products.csv
p1 und p2 gehören zu Office; p3 zu Home. Ein Join über customer_id wäre für diese Tabelle die falsche Beziehung. Behalten Sie die zwischenzeitlich angereicherten Bestellfelder und wählen Sie aus der Produktreferenz nur category.
Diese vollständige Version erhält den ersten Join und ergänzt den zweiten. Die ausdrückliche Projektion behält eine product_id, statt zwei gleichnamige Schlüssel stehen zu lassen. Anschließend berechnen das bekannte groupBy und die Aggregatfunktion sum den Kategorieumsatz:
cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("EnrichOrders").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")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")
with_customers = (
paid.alias("o")
.join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
.select(
col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
col("c.region").alias("region"),
)
)
enriched = (
with_customers.alias("e")
.join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
.select(
col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
col("e.region").alias("region"), col("p.category").alias("category"),
)
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/product-join.txt
cat /home/labex/project/output/product-join.txt
Das angereicherte Ergebnis sollte weiterhin vier bezahlte Bestellungen enthalten, nun mit Region und Kategorie. Der Kategoriebericht sollte Home 30.00 und Office 95.00 zeigen. Ihre Summe ist derselbe bezahlte Umsatz von 125.00.
Die bei vier bleibende Zeilenanzahl und der abgeglichene Umsatz liefern zwei nützliche Prüfungen. Keiner von beiden sollte sich allein durch beschreibende Spalten erhöhen. Die Schlüssel dieses Beispiels sind eindeutig und vollständig; der nächste Schritt untersucht eine fehlende Zuordnung separat, statt die Hauptverkaufssummen zu ändern.
Vergleichen Sie INNER und LEFT JOIN bei einem fehlenden Kunden
In diesem Schritt beobachten Sie das Verhalten jedes Join-Typs, wenn die Referenz keine Zuordnung findet. Prüfen Sie das unabhängige Beispiel:
cat /home/labex/project/data/example-orders.csv
demo1 verweist auf c1, demo2 auf c_missing, das in customers fehlt. Sagen Sie zuerst INNER JOIN voraus: Nur demo1 sollte bleiben. LEFT JOIN behält jeden Beispieldatensatz der linken Seite und liefert NULL für Felder, die aus einem fehlenden rechten Datensatz stammen würden.
Speichern Sie ein kleines separates Programm, damit das Hauptskript für angereicherte Verkäufe und seine Daten unverändert bleiben:
cat > /home/labex/project/compare_joins.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("CompareCustomerMatches").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
examples = spark.read.option("header", True).schema("order_id string, customer_id string").csv("/home/labex/project/data/example-orders.csv")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
inner = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
left = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "left").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
print("Inner matches:")
inner.orderBy("order_id").show(truncate=False)
print("Left matches:")
left.orderBy("order_id").show(truncate=False)
inner.write.mode("overwrite").parquet("/home/labex/project/output/example-inner")
left.write.mode("overwrite").parquet("/home/labex/project/output/example-left")
spark.stop()
PYTHON
bigdata-python /home/labex/project/compare_joins.py > /home/labex/project/output/join-comparison.txt
cat /home/labex/project/output/join-comparison.txt
Inner matches sollte demo1/c1/East enthalten. Left matches sollte diese Zeile und zusätzlich demo2/c_missing/NULL enthalten. Die customer_id der linken Seite bleibt verfügbar, obwohl die Region der rechten Seite fehlt. NULL ist das Fehlen einer zugeordneten Region und keine Region mit dem wörtlichen Namen NULL.
INNER JOIN kann für einen Bericht geeignet sein, dessen vollständige Zuordnungen bereits feststehen. LEFT JOIN kann ungeklärte Datensätze für Qualitätsprüfungen oder eine bewusste Geschäftsregel erhalten. Die Wahl des Join-Typs ändert, welche Datensätze übrig bleiben; sie ist keine bloße Anzeigepräferenz.
Verwenden Sie die angereicherten Felder für einen Regionsvergleich
In diesem Schritt wenden Sie bereits gelernte Filterung und Gruppierung auf den angereicherten Hauptdatensatz an. Der regionale Umsatz über alle Kategorien sollte East 70.00 und West 55.00 sein. Sagen Sie das reine Office-Ergebnis voraus: East bleibt 70.00, während West 25.00 wird, weil seine Home-Bestellung über 30.00 ausgeschlossen wird.
Führen Sie das vollständige Hauptskript mit zwei zusätzlichen abgeleiteten Berichten aus. Seine Joins bleiben gleich; die Kategoriebedingung ändert nur, welche angereicherten Bestellzeilen in die zweite Aggregation eingehen:
cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("EnrichOrders").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")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")
with_customers = (
paid.alias("o")
.join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
.select(
col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
col("c.region").alias("region"),
)
)
enriched = (
with_customers.alias("e")
.join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
.select(
col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
col("e.region").alias("region"), col("p.category").alias("category"),
)
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
regional = enriched.groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
office = enriched.filter(col("category") == "Office").groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
regional.write.mode("overwrite").parquet("/home/labex/project/output/regional-sales")
office.write.mode("overwrite").parquet("/home/labex/project/output/office-regional-sales")
print("All categories by region:")
regional.orderBy("region").show(truncate=False)
print("Office category by region:")
office.orderBy("region").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/regional-comparison.txt
cat /home/labex/project/output/regional-comparison.txt
Erwarten Sie die vorhergesagten regionalen Ergebnisse für alle Kategorien und nur Office. Die Office-Regionen ergeben zusammen 95.00, passend zur Office-Kategoriezwischensumme aus dem vorherigen Schritt. Alle Regionen zusammen ergeben 125.00.
Diese Abgleiche erklären sowohl die Join-Beziehung als auch die ausgewählte fachliche Teilmenge. Ihr angereicherter Parquet-Hauptdatensatz enthält weiterhin die vier korrekt typisierten bezahlten Bestellungen, und die Demonstration fehlender Zuordnungen bleibt separat. Als Nächstes drücken Sie vergleichbare Operationen mit Spark SQL über einer DataFrame-Ansicht aus.
Zusammenfassung
Sie haben Bestellungen über ausdrückliche eindeutige Kunden- und Produktschlüssel angereichert, eindeutige Ausgabespalten ausgewählt und Kategorie- und Regionssummen abgeglichen. INNER und LEFT JOIN zeigten bewusst unterschiedliches Verhalten bei fehlenden Zuordnungen. Als Nächstes vergleichen Sie Spark SQL und DataFrame-Ausdrücke auf denselben strukturierten Daten.
