Einführung
Manche Systeme liefern Datensätze als CSV-Zeilen, andere als JSON-Objekte mit benannten Feldern. Der Einzelhändler stellt nun Ereignisdateien mit einer Typbezeichnung und bestellbezogenen Werten bereit. Ihre Verarbeitungsfragen sind bekannt: Was bedeuten die Felder, welche Datensätze gehören zu Bestellungen und welche Spalten soll das Ergebnis bereitstellen?
JSON Lines speichert ein vollständiges JSON-Objekt pro Zeile. Anders als bei CSV gehören Feldnamen zu jedem Objekt, und ein Objekt kann ein weiteres Objekt enthalten. Sie lesen zunächst flache Datensätze mit ausdrücklichen Typen, wählen dann Felder aus einem einfachen verschachtelten Kundenobjekt und wenden schließlich dieses Schema auf eine zweite Lieferung an.
Dies sind endliche Dateien, die mit spark.read gelesen werden; es handelt sich weiterhin um Batch-Verarbeitung. Ein Dateiname mit events macht eine Anwendung nicht zu Streaming. Umgebung und Beispieldaten sind unabhängig; Sie benötigen die bisherigen Konzepte ausdrückliches Schema, select, filter und alias sowie grundlegendes Python. Kafka, Hadoop, Hive, Streaming-Einrichtung, Arrays oder komplexe Rekursion sind nicht erforderlich. Verwenden Sie das Terminal.
Lesen Sie flache JSON Lines und wählen Sie Bestellereignisse aus
In diesem Schritt übertragen Sie das Einlesen mit ausdrücklichem Schema von CSV auf JSON. Prüfen Sie die bereitgestellte Datei:
cd /home/labex/project
cat /home/labex/project/data/flat-events.jsonl
Jede Zeile ist ein Objekt in geschweiften Klammern. Namen in doppelten Anführungszeichen kennzeichnen Felder; ein Doppelpunkt trennt Namen und Wert. Zahlen wie quantity sind numerische JSON-Werte, während order_date eine Datumszeichenfolge in Anführungszeichen ist, die der deklarierte DateType interpretiert. Es gibt weder eine CSV-Kopfzeile noch ein umgebendes Array.
order_id, Menge, Preis und Datum des ersten Objekts beschreiben dieselbe Art von Werten wie eine CSV-Bestellung. Die Datei enthält auch ein inventory-Ereignis; einen gültigen Datensatz einzulesen bedeutet nicht, dass er in ein Bestellergebnis gehört. Das ausdrückliche Schema deklariert die flachen Felder. json(path) liest dieses Format, während das bekannte filter event_type == order auswählt.
Speichern Sie das vollständige Skript mit einem Here-Dokument in Anführungszeichen und führen Sie es mit dem vorbereiteten PySpark-Interpreter aus:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/flat-events.jsonl")
)
events.printSchema()
events.orderBy("order_id").show(truncate=False)
orders = events.filter(col("event_type") == "order")
print("Order events:")
orders.select("order_id", "quantity", "unit_price", "order_date").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/flat-json-orders.txt
cat /home/labex/project/output/flat-json-orders.txt
Die erste Anzeige sollte alle drei Ereignisdatensätze enthalten. Unter Order events erwarten Sie nur j1/2/10.00/2026-10-14 und j2/1/25.00/2026-10-14. Der inventory-Datensatz wurde erfolgreich gelesen, aber durch die Typbedingung ausgeschlossen. Das Schema sollte eine ganzzahlige Menge, einen Dezimalpreis und order_date vom Typ date zeigen.
Der Reader verwendet die benannten JSON-Felder mit den deklarierten Typen. Anders als bei einem einfachen CSV-Datensatz bestimmt die Feldreihenfolge innerhalb jedes JSON-Objekts nicht die Identität der Spalte. Das Schema definiert weiterhin die vorgesehene Datenvereinbarung.
Wählen Sie Felder aus einem verschachtelten Kundenobjekt
In diesem Schritt lesen Sie eine zusätzliche Strukturebene. Prüfen Sie die Kundenereignisdatei:
cat /home/labex/project/data/customer-events.jsonl
Der Wert customer ist ein weiteres Objekt mit customer_id und region. Im Spark-Schema wird daraus ein struct, ein Feld mit benannten untergeordneten Feldern. Es ist weder ein Array noch eine zweite Tabelle für einen Join.
Deklarieren Sie dieses struct mit einem inneren StructType. col("customer.region") wählt das untergeordnete Feld über seinen punktgetrennten Pfad; alias("region") gibt dem ausgewählten Ergebnis einen einfachen Spaltennamen. Wenden Sie dieselbe Idee auf die Kundenkennung an:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/nested-json-orders.txt
cat /home/labex/project/output/nested-json-orders.txt
Das vollständige Schema sollte customer als struct mit zwei untergeordneten Textfeldern zeigen. Die ausgewählte Tabelle sollte j3/c1/East/2/10.00/2026-10-15 und j4/c2/West/1/25.00/2026-10-15 enthalten. Ihr abschließendes Schema stellt customer_id und region als gewöhnliche Textspalten neben Menge, Preis und Datum bereit.
Sie haben Werte ausgewählt, die bereits in jedem Bestellereignis enthalten waren. Eine Nachschlagetabelle oder ein Join war nicht nötig. Die ursprüngliche verschachtelte Darstellung bleibt in events verfügbar, während selected nachfolgenden Operationen eine einfachere flache Ansicht liefert.
Verwenden Sie das verschachtelte Schema für eine weitere Datei
In diesem Schritt ändern Sie die Lieferung bei unveränderter Struktur. Lesen Sie zuerst die neue Datei:
cat /home/labex/project/data/new-customer-events.jsonl
Sagen Sie die Ausgabe voraus: eine Bestellung j5 für c2 in West, Menge 3 und Preis 12.50 am 16. Oktober. Reader und verschachtelte Auswahl können gleich bleiben; nur der Eingabepfad ändert sich. Führen Sie diese vollständige Version aus:
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/new-customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/new-json-orders.txt
cat /home/labex/project/output/new-json-orders.txt
Erwarten Sie das vorhergesagte Ergebnis mit einer Zeile und denselben ausgewählten Typen. Eine neue Datei erfordert kein neues Schema, wenn die vereinbarte Struktur unverändert ist.
Alle drei Lesevorgänge verwendeten spark.read auf vorhandenen Dateien und endeten mit ihren Skripten. Sie abonnierten keine künftigen Nachrichten. Sie haben gelernt, ein Eingabeformat zu variieren, ohne gleichzeitig das Ausführungsmodell zu ändern. Als Nächstes kehren Sie zu einer bewusst unvollkommenen Bestelllieferung zurück und definieren ausdrückliche Bereinigungsregeln.
Zusammenfassung
Sie haben statische JSON Lines mit ausdrücklichen skalaren Typen gelesen, Bestellereignisse ausgewählt und eine einfache Kundenstruktur mit benannten Spaltenpfaden geglättet. Eine zweite Lieferung verwendete dasselbe Schema korrekt. Als Nächstes behandeln Sie fehlende, doppelte und ungültige Bestelldatensätze nach klaren Bereinigungsregeln.
