Introduction
A script can run successfully and still pass bad records to a report. Today's retail delivery contains a repeated order row, a missing customer, invalid numeric text, a nonpositive quantity, a negative price, and a malformed date. Automatically filling these problems with zero would invent business information.
You will separate three decisions: read text without losing the original evidence, safely parse intended numeric/date fields, and retain only rows meeting a stated rule. Identical valid rows are then deduplicated. Finally, you will save and read back a clean Parquet dataset. Parquet is a typed binary columnar file format; this lab teaches only the minimal write/read workflow, with partitioned storage reserved for later.
The prepared environment supplies two independent deliveries. You should know CSV reading, Spark column expressions, filtering, and decimal prices. The safe conversion expression used here is supported by the prepared Spark version; you do not need to change its error-handling configuration. Use the Terminal. No previous output files, Hadoop, Hive, or service installation is required.
Preserve the Delivery as String Staging Data
In this step, you will observe defects before converting values. Inspect the local source:
cd /home/labex/project
cat /home/labex/project/data/dirty-orders.csv
There are fourteen business rows. A staging DataFrame keeps these input fields as strings, allowing you to see values such as two, unknown, and not-a-date without pretending they are valid numbers or dates. The CSV reader's default handling makes an empty customer field NULL; NULL represents a missing value, not the text "NULL".
Save and run the complete staging program. As before, the quoted PYTHON here-document writes the source literally. show(30, truncate=False) permits up to thirty displayed records and keeps their field text readable:
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
Expect fourteen displayed records and Staging rows: 14. All columns should be strings. Find b2's quantity two and b4's price unknown; these are useful evidence rather than errors hidden by a guessed default. o2 appears twice, and o1's status has surrounding spaces.
The original file is read-only in this exercise. You will create derived DataFrames and output directories while leaving the raw delivery available for review.
Parse Values Safely Without Inventing Defaults
In this step, you will normalize the two text fields that need surrounding whitespace removed and attempt numeric/date conversion. trim removes leading and trailing spaces. It does not guess a missing customer or change an order's payment status.
expr lets you supply a built-in Spark expression as text. try_cast(quantity AS INT) attempts a whole-number conversion and produces NULL if this value cannot be converted. The equivalent expressions request DECIMAL(10,2) for prices and DATE for dates. Ordinary failing casts may raise errors under Spark's ANSI behavior; this explicit safe conversion avoids relying on a global setting.
The schema change preserves rows so that parsing problems remain observable. Run this version:
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
Expect Parsed rows: 14. b2's quantity, b4's price, and b6's date should now be NULL. b3's zero quantity and b5's negative price are valid numeric values syntactically, but they still fail the business rules you will apply next. o1's status should now be paid without surrounding spaces.
Successful conversion is not the same as a valid order. Keeping these stages distinct lets you explain whether a row failed parsing or a business constraint. No missing price was replaced with zero; the real zero-priced order o7 remains distinguishable from an unknown price.
Apply Explicit Validity and Duplicate Rules
In this step, you will keep valid business records and remove only identical duplicates. Use these rules:
- customer_id is present and nonempty after trimming;
- quantity parsed successfully and is greater than zero;
- unit_price parsed successfully and is at least zero;
- order_date parsed successfully;
- rows identical across all seven normalized fields are retained once.
The fixture's other identifier fields are already well formed. It has no conflicting versions of the same order ID, so this exercise does not choose a winner among changing statuses. Paid, pending, and refunded records all remain if otherwise valid; cleaning a delivery is different from choosing paid revenue for a report.
isNotNull explicitly checks presence. A comparison involving NULL does not become a true match, and filter retains only true conditions. Combining the conditions with & makes the policy visible. dropDuplicates() with no subset removes identical full rows; using only order_id would be a different policy that could discard conflicting information.
Run this version:
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
Expect Staging count: 14, Valid count before deduplication: 8, and Clean count: 7. b1–b6 are excluded by the stated rules, while one repeated o2 is removed. The seven retained IDs are o1 through o7.
Order o7's real zero price is valid under the nonnegative-price rule. o3 is still pending and o5 refunded: neither status makes the source record malformed. The resulting dataset is clean for this stated policy, not a claim that every possible real-world anomaly has been solved.
Save, Read Back, and Test One Extra Defect
In this step, you will make the clean dataset reusable and test the same policy on another delivery. write.mode("overwrite").parquet(path) writes a Parquet dataset to a directory, replacing previous derived contents there. Keep that directory separate from the raw input. Spark can create multiple part files and a completion marker; do not depend on a particular file name or count.
spark.read.parquet(path) reads the saved dataset and its stored types. Querying that read-back DataFrame checks the actual artifact, rather than merely displaying the pre-write plan. The full program below places the already learned logic in a Python function so both deliveries use the same cleaning rule.
def clean_orders(input_path): defines the function and names its input parameter. The indented lines are its body; they run when the function is called. The first call supplies the original filename, and the second supplies the extra-delivery filename. Inside each call, input_path refers to that supplied value. return clean gives the resulting DataFrame back to the caller; it does not save files. The unindented writer lines perform that separate action. Preserve the indentation when copying the complete program.
The extra delivery contains the original fourteen rows plus a negative-quantity row and another exact copy of the valid zero-priced order. Predict the clean result: those additions should not change the seven retained records. Inspect the additions:
tail -n 2 /home/labex/project/data/dirty-orders-extra.csv
Run the complete save-and-read-back program:
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
Both read-back tables should contain o1 through o7 exactly once, and both counts should be 7. The main output directory is output/clean-orders; practice-clean-orders is a separate derived experiment. The read-back schema should retain integer quantity, decimal price, and date order_date.
This is a minimal Parquet round trip, not a speed comparison or storage-tuning exercise. You now have a small clean dataset another Spark process can read, and evidence that a newly introduced invalid row and duplicate do not silently change the result.
Summary
You preserved raw text, used safe conversion to expose invalid values, applied an explicit missing-value and range policy, and removed identical duplicate rows. You retained valid zero prices and nonpaid records, then verified the actual typed Parquet output and a second delivery. The next challenge lets you apply these cleaning and amount-calculation skills independently.
