Einführung
Ein Skript kann erfolgreich laufen und trotzdem schlechte Datensätze an einen Bericht weitergeben. Die heutige Handelslieferung enthält eine wiederholte Bestellzeile, einen fehlenden Kunden, ungültigen numerischen Text, eine nichtpositive Menge, einen negativen Preis und ein fehlerhaftes Datum. Diese Probleme automatisch mit null zu füllen würde Geschäftsinformationen erfinden.
Sie trennen drei Entscheidungen: Text lesen, ohne die ursprünglichen Belege zu verlieren, vorgesehene Zahlen- und Datumsfelder sicher konvertieren und nur Zeilen behalten, die eine festgelegte Regel erfüllen. Identische gültige Zeilen werden anschließend dedupliziert. Zum Schluss speichern Sie einen sauberen Parquet-Datensatz und lesen ihn zurück. Parquet ist ein typisiertes binäres spaltenorientiertes Dateiformat; dieses Lab behandelt nur den minimalen Schreib-/Leseablauf, partitionierte Speicherung folgt später.
Die vorbereitete Umgebung stellt zwei unabhängige Lieferungen bereit. CSV-Einlesen, Spark-Spaltenausdrücke, Filterung und Dezimalpreise sollten Sie kennen. Der hier verwendete sichere Konvertierungsausdruck wird von der vorbereiteten Spark-Version unterstützt; Sie müssen ihre Fehlerbehandlungskonfiguration nicht ändern. Verwenden Sie das Terminal. Frühere Ausgabedateien, Hadoop, Hive oder Dienstinstallation sind nicht erforderlich.
Bewahren Sie die Lieferung als Text-Staging-Daten auf
In diesem Schritt beobachten Sie Mängel vor der Konvertierung von Werten. Prüfen Sie die lokale Quelle:
cd /home/labex/project
cat /home/labex/project/data/dirty-orders.csv
Es gibt vierzehn Geschäftszeilen. Ein Staging-DataFrame hält diese Eingabefelder als Zeichenfolgen vor, damit Werte wie two, unknown und not-a-date sichtbar bleiben, ohne sie als gültige Zahlen oder Datumswerte auszugeben. Die Standardbehandlung des CSV-Readers macht ein leeres Kundenfeld zu NULL; NULL bedeutet einen fehlenden Wert, nicht den Text „NULL“.
Speichern und starten Sie das vollständige Staging-Programm. Wie zuvor schreibt das Here-Dokument mit PYTHON in Anführungszeichen den Quelltext wörtlich. show(30, truncate=False) erlaubt bis zu dreißig angezeigte Datensätze und hält ihren Feldtext lesbar:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
staging.printSchema()
staging.orderBy("order_id").show(30, truncate=False)
print("Staging rows:", staging.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/staging-orders.txt
cat /home/labex/project/output/staging-orders.txt
Erwarten Sie vierzehn angezeigte Datensätze und Staging rows: 14. Alle Spalten sollten Zeichenfolgen sein. Finden Sie die Menge two von b2 und den Preis unknown von b4; das sind nützliche Belege statt durch einen geratenen Standardwert versteckte Fehler. o2 erscheint zweimal, und der Status von o1 hat umgebende Leerzeichen.
Die Originaldatei ist in dieser Übung schreibgeschützt. Sie erstellen abgeleitete DataFrames und Ausgabeverzeichnisse, während die Rohdatenlieferung zur Prüfung verfügbar bleibt.
Konvertieren Sie Werte sicher, ohne Standardwerte zu erfinden
In diesem Schritt normalisieren Sie die zwei Textfelder, deren umgebende Leerzeichen entfernt werden sollen, und versuchen die Zahlen-/Datumskonvertierung. trim entfernt führende und abschließende Leerzeichen. Es errät keinen fehlenden Kunden und ändert keinen Zahlungsstatus einer Bestellung.
expr erlaubt, einen eingebauten Spark-Ausdruck als Text anzugeben. try_cast(quantity AS INT) versucht eine Ganzzahlkonvertierung und erzeugt NULL, wenn dieser Wert nicht konvertiert werden kann. Die entsprechenden Ausdrücke verlangen DECIMAL(10,2) für Preise und DATE für Datumswerte. Gewöhnliche fehlschlagende Konvertierungen können unter Sparks ANSI-Verhalten Fehler auslösen; diese ausdrückliche sichere Konvertierung vermeidet die Abhängigkeit von einer globalen Einstellung.
Die Schemaänderung erhält Zeilen, damit Konvertierungsprobleme beobachtbar bleiben. Führen Sie diese Version aus:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
parsed.printSchema()
parsed.orderBy("order_id").show(30, truncate=False)
print("Parsed rows:", parsed.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/parsed-orders.txt
cat /home/labex/project/output/parsed-orders.txt
Erwarten Sie Parsed rows: 14. Die Menge von b2, der Preis von b4 und das Datum von b6 sollten nun NULL sein. Die Menge null von b3 und der negative Preis von b5 sind syntaktisch gültige Zahlenwerte, verletzen aber weiterhin die als Nächstes angewandten Geschäftsregeln. Der Status von o1 sollte nun paid ohne umgebende Leerzeichen sein.
Erfolgreiche Konvertierung ist nicht dasselbe wie eine gültige Bestellung. Durch die Trennung dieser Phasen können Sie erklären, ob eine Zeile an der Konvertierung oder einer fachlichen Bedingung scheiterte. Kein fehlender Preis wurde durch null ersetzt; die tatsächlich kostenlos bepreiste Bestellung o7 bleibt von einem unbekannten Preis unterscheidbar.
Wenden Sie ausdrückliche Gültigkeits- und Duplikatregeln an
In diesem Schritt behalten Sie gültige Geschäftsdatensätze und entfernen nur identische Duplikate. Verwenden Sie diese Regeln:
- customer_id ist nach dem Entfernen umgebender Leerzeichen vorhanden und nicht leer;
- quantity wurde erfolgreich konvertiert und ist größer als null;
- unit_price wurde erfolgreich konvertiert und ist mindestens null;
- order_date wurde erfolgreich konvertiert;
- über alle sieben normalisierten Felder identische Zeilen bleiben einmal erhalten.
Die anderen Kennungsfelder der Beispieldaten sind bereits korrekt aufgebaut. Es gibt keine widersprüchlichen Versionen derselben Bestell-ID; diese Übung wählt daher keinen Gewinner unter wechselnden Statuswerten. Datensätze mit paid, pending und refunded bleiben alle erhalten, wenn sie sonst gültig sind. Eine Lieferung zu bereinigen ist etwas anderes, als bezahlten Umsatz für einen Bericht auszuwählen.
isNotNull prüft ausdrücklich das Vorhandensein. Ein Vergleich mit NULL wird nicht zu einem wahren Treffer, und filter behält nur wahre Bedingungen. Das Kombinieren mit & macht die Regel sichtbar. dropDuplicates() ohne Teilmenge entfernt identische vollständige Zeilen; nur order_id zu verwenden wäre eine andere Regel, die widersprüchliche Informationen verwerfen könnte.
Führen Sie diese Version aus:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
print("Clean rows:")
clean.orderBy("order_id").show(30, truncate=False)
print("Staging count:", staging.count())
print("Valid count before deduplication:", valid.count())
print("Clean count:", clean.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/cleaning-result.txt
cat /home/labex/project/output/cleaning-result.txt
Erwarten Sie Staging count: 14, Valid count before deduplication: 8 und Clean count: 7. b1–b6 werden durch die festgelegten Regeln ausgeschlossen, während ein wiederholtes o2 entfernt wird. Die sieben erhaltenen IDs sind o1 bis o7.
Der echte Nullpreis von Bestellung o7 ist unter der Regel für nichtnegative Preise gültig. o3 ist weiterhin pending und o5 refunded: Keiner dieser Statuswerte macht den Quelldatensatz fehlerhaft. Der entstandene Datensatz ist für diese festgelegte Regel sauber; das bedeutet nicht, dass jede denkbare reale Anomalie gelöst wurde.
Speichern, lesen und testen Sie einen zusätzlichen Mangel
In diesem Schritt machen Sie den sauberen Datensatz wiederverwendbar und testen dieselbe Regel mit einer weiteren Lieferung. write.mode("overwrite").parquet(path) schreibt einen Parquet-Datensatz in ein Verzeichnis und ersetzt dort frühere abgeleitete Inhalte. Halten Sie dieses Verzeichnis von der Rohdateneingabe getrennt. Spark kann mehrere part-Dateien und einen Abschlussmarker erstellen; verlassen Sie sich nicht auf bestimmte Dateinamen oder Anzahlen.
spark.read.parquet(path) liest den gespeicherten Datensatz und seine gespeicherten Typen. Die Abfrage dieses zurückgelesenen DataFrames prüft die tatsächliche Dateiablage statt nur den Plan vor dem Schreiben anzuzeigen. Das vollständige Programm legt die bereits gelernte Logik in eine Python-Funktion, damit beide Lieferungen dieselbe Bereinigungsregel verwenden.
def clean_orders(input_path): definiert die Funktion und benennt ihren Eingabeparameter. Die eingerückten Zeilen sind ihr Rumpf; sie laufen beim Funktionsaufruf. Der erste Aufruf liefert den ursprünglichen Dateinamen, der zweite den Namen der Zusatzlieferung. In jedem Aufruf verweist input_path auf den übergebenen Wert. return clean gibt den entstandenen DataFrame an den Aufrufer zurück; es speichert keine Dateien. Die nicht eingerückten Schreibzeilen führen diese separate Aktion aus. Erhalten Sie die Einrückung beim Kopieren des vollständigen Programms.
Die Zusatzlieferung enthält die ursprünglichen vierzehn Zeilen plus eine Zeile mit negativer Menge und eine weitere exakte Kopie der gültigen Bestellung mit Nullpreis. Sagen Sie das bereinigte Ergebnis voraus: Diese Ergänzungen sollten die sieben erhaltenen Datensätze nicht verändern. Prüfen Sie die Ergänzungen:
tail -n 2 /home/labex/project/data/dirty-orders-extra.csv
Führen Sie das vollständige Programm zum Speichern und Zurücklesen aus:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
def clean_orders(input_path):
staging = spark.read.option("header", True).csv(input_path)
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
return clean
clean = clean_orders("/home/labex/project/data/dirty-orders.csv")
clean.write.mode("overwrite").parquet("/home/labex/project/output/clean-orders")
reloaded = spark.read.parquet("/home/labex/project/output/clean-orders")
print("Read-back clean rows:")
reloaded.orderBy("order_id").show(30, truncate=False)
reloaded.printSchema()
print("Read-back count:", reloaded.count())
practice = clean_orders("/home/labex/project/data/dirty-orders-extra.csv")
practice.write.mode("overwrite").parquet("/home/labex/project/output/practice-clean-orders")
practice_reloaded = spark.read.parquet("/home/labex/project/output/practice-clean-orders")
print("Extra-delivery clean rows:")
practice_reloaded.orderBy("order_id").show(30, truncate=False)
print("Extra-delivery count:", practice_reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/clean-parquet-check.txt
cat /home/labex/project/output/clean-parquet-check.txt
Beide zurückgelesenen Tabellen sollten o1 bis o7 genau einmal enthalten, und beide Anzahlen sollten 7 sein. Das Hauptausgabeverzeichnis ist output/clean-orders; practice-clean-orders ist ein separater abgeleiteter Versuch. Das zurückgelesene Schema sollte ganzzahlige Menge, Dezimalpreis und order_date vom Typ date erhalten.
Dies ist ein minimaler Parquet-Schreib-/Leseablauf, kein Geschwindigkeitsvergleich oder Speichertuning. Sie haben nun einen kleinen sauberen Datensatz, den ein weiterer Spark-Prozess lesen kann, und Belege, dass eine neu eingeführte ungültige Zeile und ein Duplikat das Ergebnis nicht unbemerkt verändern.
Zusammenfassung
Sie haben Rohtext erhalten, mit sicherer Konvertierung ungültige Werte sichtbar gemacht, eine ausdrückliche Regel für fehlende Werte und Wertebereiche angewandt und identische doppelte Zeilen entfernt. Sie haben gültige Nullpreise und unbezahlte Datensätze behalten und anschließend die tatsächliche typisierte Parquet-Ausgabe sowie eine zweite Lieferung geprüft. In der nächsten Challenge wenden Sie Bereinigung und Betragsberechnung eigenständig an.
