如何对PySpark DataFrame中的字段值执行断言校验

错误原因说明 #

你之前的写法无法运行的核心原因有两个：

1. orderlines['Price'] 是PySpark的Column类型对象，直接和数值做比较返回的是列表达式，不是布尔值，不能直接用于assert判断
2. collect()返回的是Row对象组成的列表，无法直接和单个数值做大小比较

符合PySpark开发规范的实现方式 #

场景1：单规则快速校验（如检查所有Price为正）

最简洁且适配大数据量场景的写法是：先筛选出所有不符合规则的行，判断其数量是否为0即可，校验逻辑全部在Spark分布式层执行，不会把全量数据拉到Driver端引发内存溢出。

from pyspark.sql.functions import col

# 筛选Price<=0的异常行，统计数量
invalid_count = orderlines.filter(col("Price") <= 0.0).count()
assert invalid_count == 0, f"发现{invalid_count}条Price为非正数的异常数据"

场景2：多规则批量校验

如果同时需要校验多个业务规则（比如同时检查Amount为正、OrderID不为空等），可以统一构造校验结果列，批量统计异常：

from pyspark.sql.functions import col, when, lit

# 构造每一条数据的校验结果列，False代表校验不通过
orderlines_with_check = orderlines.withColumn(
    "is_price_valid",
    when(col("Price") > 0.0, lit(True)).otherwise(lit(False))
).withColumn(
    "is_amount_valid",
    when(col("Amount") > 0, lit(True)).otherwise(lit(False))
)

# 统计所有校验不通过的总行数
total_invalid = orderlines_with_check.filter(
    col("is_price_valid") == False | col("is_amount_valid") == False
).count()
assert total_invalid == 0, f"共发现{total_invalid}条不符合校验规则的数据"

内容的提问来源于stack exchange，提问作者Larsq