使用PySpark实现DataFrame列0~100及1~5区间取值校验的方法

该方案可一次性完成多列范围校验，仅扫描一次DataFrame，性能优异，校验不通过时会抛出包含所有错误详情的异常并终止运行。

实现步骤 #

• 导入PySpark相关依赖函数
• 自定义校验规则，支持灵活扩展多列校验
• 全量扫描一次DataFrame，统计所有列的非法数据数量
• 汇总校验结果，存在非法数据时直接抛出异常终止程序

完整示例代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, when

# 初始化SparkSession
spark = SparkSession.builder.appName("DataRangeValidation").getOrCreate()

# 替换为你自己的DataFrame数据源，支持csv/parquet/hive表等任意来源
df = spark.read.csv("your_data_path.csv", header=True, inferSchema=True)

# 校验规则配置：(列名, 合法最小值, 合法最大值, 错误提示前缀)
# 可按需新增更多列的校验规则
validation_rules = [
    ("column_0_100", 0, 100, "列column_0_100存在超出0~100范围的非法值，非法行数："),
    ("column_1_5", 1, 5, "列column_1_5存在超出1~5范围的非法值，非法行数：")
]

# 统计所有列的非法数据量，仅扫描一次DataFrame
invalid_result = df.select([
    count(when(~col(col_name).between(min_val, max_val), 1)).alias(f"invalid_{col_name}")
    for col_name, min_val, max_val, _ in validation_rules
]).first()

# 汇总错误信息
error_list = []
for index, (col_name, _, _, err_prefix) in enumerate(validation_rules):
    invalid_num = invalid_result[index]
    if invalid_num > 0:
        error_list.append(f"{err_prefix}{invalid_num}")

# 校验不通过则抛出异常终止运行
if error_list:
    raise ValueError(f"数据校验失败，错误详情：\n{'\n'.join(error_list)}")

# 校验通过后执行后续业务逻辑
print("所有数据校验通过")

注意事项 #

• 如果待校验列不是数值类型，需要先使用col(col_name).cast("int")（或float类型）转换后再做校验，避免逻辑错误
• 该方案性能远高于逐列单独校验，适合TB级以上大数据量场景
• 若需要留存非法行用于排查问题，可在统计非法量的同时过滤出非法行写入指定存储路径，再抛出异常

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