Einführung
Die Bestelldatensätze des Einzelhändlers kommen nun als CSV-Datei an, statt direkt in Ihre Python-Liste geschrieben zu werden. Eine Datei speichert Text wie 2 und 10.00; eine Berechnung muss wissen, ob diese Werte Text, eine ganzzahlige Menge oder einen Dezimalpreis bedeuten. Die richtigen Zeichen zu lesen ist nur ein Teil des korrekten Einlesens von Daten.
Sie vergleichen Sparks standardmäßiges Laden als Text, automatische Typerkennung und ein ausdrückliches Schema. Anschließend wenden Sie dasselbe Schema auf eine zweite Datei mit derselben Feldreihenfolge an. Die Geschäftsdaten sind hier bewusst gültig, damit Sie sich auf die Vereinbarung zwischen Spalten und Werten konzentrieren können. Ein späteres Lab behandelt fehlende und ungültige Eingaben ausdrücklich.
Die Umgebung stellt PySpark und zwei unabhängige lokale CSV-Beispiele bereit. SparkSession, DataFrame-Anzeige, Schema und Zeilenanzahl aus dem ersten Lab sowie grundlegendes Python sollten Sie verstehen. pandas, Hadoop, Hive, Installation oder vorheriger VM-Zustand sind nicht erforderlich. Verwenden Sie das Terminal.
Lesen Sie die Kopfzeile und beobachten Sie Textspalten
In diesem Schritt verbinden Sie die CSV-Feldpositionen mit der von Spark erstellten Tabelle. Prüfen Sie zuerst die tatsächliche Datei:
cd /home/labex/project
cat /home/labex/project/data/orders.csv
Die erste Zeile benennt sieben Spalten; die nächsten sechs Zeilen sind Bestellungen. Jede Zeile stellt eine Bestellung mit einem Produkttyp dar. Die Datei enthält die Statuswerte paid, pending und refunded; dieses Lab liest sie alle, ohne eine Verkaufsregel anzuwenden.
spark.read erstellt einen DataFrame-Reader. option("header", True) sagt diesem Reader, die erste CSV-Zeile als Spaltennamen statt als Bestelldatensatz zu behandeln. csv(input_path) benennt die lokale Datei. Ohne Typerkennung oder bereitgestelltes Schema erzeugt der CSV-Reader Textspalten.
Speichern und starten Sie dieses vollständige Programm. Das Here-Dokument mit dem Marker PYTHON in Anführungszeichen erhält den eingeschlossenen Quelltext, und > schreibt die lokale Datei. bigdata-python verwendet die vorbereitete PySpark-Umgebung:
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/orders.csv"
df = spark.read.option("header", True).csv(input_path)
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/string-orders.txt
cat /home/labex/project/output/string-orders.txt
Erwarten Sie sechs Zeilen und sieben Textspalten, einschließlich quantity und unit_price. Die angezeigte 2 kann numerisch aussehen, aber printSchema kennzeichnet ihren tatsächlichen Typ string. orderBy("order_id") sortiert die angezeigten Zeilen nach Kennung für eine vorhersehbare Prüfung; Sortieren ist von der Typzuweisung getrennt.
header=True verhinderte, dass die Kopfzeile zu einem siebten Datensatz wurde. Es sagte Spark nicht, dass Preise exakte Dezimalzahlen sein oder Datumswerte Datumsoperationen unterstützen sollen.
Vergleichen Sie Typerkennung und eine ausdrückliche Datenvereinbarung
In diesem Schritt deklarieren Sie die Bedeutung jedes CSV-Felds. inferSchema=True fordert Spark auf, Werte zu untersuchen und Typen abzuleiten. Das kann bequem sein, doch das Ergebnis hängt von den beobachteten Daten ab; es ist keine dauerhafte Vereinbarung über Preise oder künftige Dateien. In diesem Beispiel erkennt die Typerkennung unit_price als double, also eine Gleitkommazahl.
Verwenden Sie ein ausdrückliches Schema für die vorgesehene Vereinbarung. StructType ist die geordnete Feldsammlung. Jedes StructField enthält Namen, Typ und True, um fehlende Werte zuzulassen. StringType enthält Kennungen und Statustext; IntegerType Mengen; DecimalType(10, 2) exakte Preise mit insgesamt zehn Stellen und zwei Nachkommastellen; DateType Kalenderdaten. Nullable-Felder erlauben fehlende Werte, machen fehlende Geschäftsinformationen aber nicht automatisch akzeptabel.
Ersetzen Sie das Programm durch diese Version. Sie gibt die erkannte Typbeschreibung zum Vergleich aus und liest dann dieselbe Datei mit Ihrem ausdrücklichen Schema:
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").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),
])
input_path = "/home/labex/project/data/orders.csv"
inferred = spark.read.option("header", True).option("inferSchema", True).csv(input_path)
print("Inferred schema:", inferred.schema.simpleString())
df = (
spark.read
.option("header", True)
.option("dateFormat", "yyyy-MM-dd")
.schema(order_schema)
.csv(input_path)
)
print("Explicit schema:")
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/typed-orders.txt
cat /home/labex/project/output/typed-orders.txt
Finden Sie das ausdrückliche Schema unterhalb der Zusammenfassung der Typerkennung. quantity sollte integer, unit_price decimal(10,2) und order_date date sein. Alle sechs Datensätze sollten weiterhin vorhanden sein. Die ausdrückliche Option dateFormat beschreibt die Eingabedaten als Jahr-Monat-Tag, passend zu Werten wie 2026-10-14.
Die Feldreihenfolge des Schemas stimmt mit der dieser CSV überein. Ein bereitgestelltes Schema ist keine allgemeine Anweisung, jede eingehende Datei anhand ihrer Kopfzeilennamen umzuordnen. Für diese Lieferung mit bekannter Reihenfolge macht die Vereinbarung die vorgesehenen Spalten und Typen klar. Ein Schema ersetzt auch keine Datenqualitätsregel; später lernen Sie den sicheren Umgang mit ungültigen Werten.
Wenden Sie dasselbe Schema auf eine neue Lieferung an
In diesem Schritt testen Sie die Vereinbarung an einer zweiten gültigen Datei. Prüfen Sie sie zuerst:
cat /home/labex/project/data/new-orders.csv
Sie hat dieselben sieben Felder in derselben Reihenfolge und zwei neue Bestellungen. Sagen Sie die Änderungen voraus: zwei Geschäftszeilen statt sechs, während Menge, Preis und Datum ihre deklarierten Typen behalten.
Die folgende vollständige Version behält das ausdrückliche Schema bei und ändert den Eingabepfad. Der Vergleich der Typerkennung ist für dieses normale Einlesen nicht mehr erforderlich:
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").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),
])
input_path = "/home/labex/project/data/new-orders.csv"
df = (
spark.read
.option("header", True)
.option("dateFormat", "yyyy-MM-dd")
.schema(order_schema)
.csv(input_path)
)
print("Explicit schema:")
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/new-typed-orders.txt
cat /home/labex/project/output/new-typed-orders.txt
Erwarten Sie o7 mit Menge 2, Preis 12.50, Status paid und Datum 2026-10-16; o8 mit Menge 3, Preis 4.00, Status pending und Datum 2026-10-17. Die Anzahl sollte 2 sein. Prüfen Sie dieselben Ganzzahl-, Dezimal- und Datumstypen in printSchema.
Die ursprüngliche Datei mit sechs Bestellungen bleibt unverändert. Ein wiederverwendbares Schema erlaubt einer neuen Lieferung, dieselbe Verarbeitungslogik zu durchlaufen, ohne die Bedeutung von Menge oder Preis neu zu definieren. Als Nächstes verwenden Sie diese typisierten Spalten zum Auswählen von Bestellungen und zum Berechnen des Betrags jeder Bestellung.
Zusammenfassung
Sie haben eine CSV-Kopfzeile korrekt gelesen, Textladen mit Typerkennung verglichen und eine ausdrückliche Vereinbarung für Kennungen, Mengen, Dezimalpreise und Datumswerte deklariert. Dasselbe Schema interpretierte eine zweite Lieferung korrekt. Als Nächstes können Sie mit typisierten Spaltenausdrücken Zeilen filtern und Bestellbeträge berechnen.
