清洗缺失、重复和无效的 Spark 订单记录

介绍

脚本可以成功运行，却仍把坏记录传给报表。今天收到的零售数据包含重复订单行、缺失客户、无效数值文本、非正数量、负价格和格式错误的日期。自动把这些问题填成零，会虚构业务信息。

你将区分三项决策：读取文本时保留原始证据，安全解析预期数值和日期字段，以及只保留符合明确规则的行。随后对相同的有效行去重。最后，保存并读回干净 Parquet 数据集。Parquet 是具有类型的二进制列式文件格式；本实验只教最小写入/读取流程，分区存储留待后续。

预配置环境提供两批独立数据。你应该了解 CSV 读取、Spark 列表达式、筛选和十进制价格。这里使用的安全转换表达式受预配置 Spark 版本支持，无需更改错误处理配置。请使用终端。无需之前的输出文件、Hadoop、Hive 或安装服务。

将交付数据保留为字符串暂存数据

在这一步中，你将在转换值之前观察缺陷。查看本地源文件：

cd /home/labex/project

cat /home/labex/project/data/dirty-orders.csv

其中有十四条业务行。暂存（staging）DataFrame 将输入字段保留为字符串，让你看到 two、unknown 和 not-a-date 等值，而不假装它们是有效数字或日期。CSV 读取器的默认处理将空客户字段变为 NULL；NULL 表示缺失值，不是文本“NULL”。

保存并运行完整暂存程序。与之前一样，带引号的 PYTHON here-document 原样写入源码。show(30, truncate=False) 最多显示三十条记录，并保留可读的字段文本：

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

预期显示十四条记录及 Staging rows: 14。所有列都应为字符串。找到 b2 的数量 two 和 b4 的价格 unknown；这些是有用证据，而不是被猜测默认值掩盖的错误。o2 出现两次，o1 的状态两侧带空格。

本练习中的原始文件只读。你将创建派生 DataFrame 和输出目录，保留原始交付数据供审核。

安全解析值，不虚构默认值

在这一步中，你将规范两个需要移除两端空白的文本字段，并尝试数值/日期转换。trim 移除开头和末尾空格，不会猜测缺失客户，也不会改变订单付款状态。

expr 允许用文本提供内置 Spark 表达式。try_cast(quantity AS INT) 尝试整数转换，无法转换时生成 NULL。对应表达式为价格请求 DECIMAL(10,2)，为日期请求 DATE。在 Spark ANSI 行为下，普通转换失败可能抛出错误；这种明确的安全转换避免依赖全局设置。

模式变化保留所有行，让解析问题仍可观察。运行这个版本：

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

预期为 Parsed rows: 14。b2 的数量、b4 的价格、b6 的日期现在应为 NULL。b3 的零数量与 b5 的负价格在语法上是有效数值，却仍然不符合接下来应用的业务规则。o1 的状态现在应为不带两端空格的 paid。

转换成功不等于订单有效。把这些阶段分开，可以解释某行是解析失败，还是不符合业务约束。没有缺失价格被替换为零；真实零价格订单 o7 仍能与未知价格区分。

应用明确的有效性与重复处理规则

在这一步中，你将保留有效业务记录，只移除完全相同的重复行。使用以下规则：

• customer_id 去除两端空格后存在且非空；
• quantity 解析成功且大于零；
• unit_price 解析成功且大于或等于零；
• order_date 解析成功；
• 在全部七个规范化字段上相同的行只保留一次。

样本中的其他标识符字段已符合格式。同一个订单 ID 不存在互相冲突的版本，因此本练习不需要在变化状态中选择一个版本。只要其他方面有效，已付款、待付款和已退款记录都保留；清洗一批数据与为报表选择已付款收入是不同操作。

isNotNull 明确检查是否存在。涉及 NULL 的比较不会成为真匹配，filter 只保留条件为真的行。用 & 组合条件，让策略清晰可见。不指定子集的 dropDuplicates() 移除完全相同的整行；只按 order_id 去重会是另一种策略，可能丢弃冲突信息。

运行这个版本：

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

预期为 Staging count: 14、Valid count before deduplication: 8 和 Clean count: 7。b1–b6 被明确规则排除，一个重复的 o2 被移除。保留的七个 ID 为 o1 到 o7。

在非负价格规则下，o7 的真实零价格有效。o3 仍为待付款，o5 仍为已退款：两种状态都不代表源记录格式错误。结果符合当前策略的清洗要求，并不表示已经解决所有现实世界异常。

保存、读回并测试额外缺陷

在这一步中，你将让干净数据集可复用，并在另一批数据上测试相同策略。write.mode("overwrite").parquet(path) 将 Parquet 数据集写入目录，替换该处已有派生内容。该目录应与原始输入分开。Spark 可以创建多个 part 文件和完成标记，不要依赖特定文件名或数量。

spark.read.parquet(path) 读取保存的数据集及其类型。查询读回的 DataFrame，检查的是实际文件成果，不只是显示写入前的计划。下面完整程序将已学逻辑放入 Python 函数，让两批数据使用相同清洗规则。

def clean_orders(input_path): 定义函数并命名输入参数。缩进行属于函数体，调用函数时才运行。第一次调用提供原始文件名，第二次提供额外数据文件名。每次调用中，input_path 都指传入的值。return clean 将结果 DataFrame 返回给调用者，不保存文件。未缩进的写入行执行这个独立操作。复制完整程序时请保留缩进。

额外交付包含原来的十四行，另外增加一个负数量行，以及一个有效零价格订单的精确副本。预测清洗结果：这些新增内容不应改变保留的七条记录。查看新增内容：

tail -n 2 /home/labex/project/data/dirty-orders-extra.csv

运行完整的保存并读回程序：

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

两张读回表都应包含 o1 到 o7，每条恰好一次，计数均为 7。主输出目录是 output/clean-orders，practice-clean-orders 是独立派生实验。读回模式应保留整数 quantity、十进制价格和 date 类型 order_date。

这是最小的 Parquet 写入读回流程，不是速度比较或存储调优练习。现在你有了另一个 Spark 进程可读取的小型干净数据集，也有证据证明新增的无效行与重复行不会悄悄改变结果。

总结

你保留原始文本，通过安全转换暴露无效值，应用明确的缺失值和范围策略，并移除完全相同的重复行。你保留有效零价格和非已付款记录，再验证实际的有类型 Parquet 输出与第二批数据。下个挑战将让你独立应用清洗和金额计算技能。