Spark中调用自定义UDF传入DataFrame参数触发TypeError的原因排查

为什么会触发TypeError？ #

你遇到的这个错误核心原因很直接：Spark UDF不允许直接把整个DataFrame作为参数传入。

UDF的设计逻辑是在分布式环境中处理DataFrame的单条数据行，它的参数只能是以下几种类型：

• 单个/多个列对象（Column）
• 字面量（比如字符串、数字，需要用F.lit()包装）

而你在调用dq_check_wrapper(df, metadata.col, metadata.dq_check)时，第一个参数传的是整个df对象，Spark无法识别这种非列/非字面量的参数类型，所以直接抛出了TypeError。

另外还有个逻辑误区：你的Valid_df类是针对整个DataFrame做列级的数据质量检查（比如判断某列是否有null、是否唯一），这类检查是全局统计性的，而UDF是逐行执行的，两者的运行逻辑完全不匹配，用UDF来包裹这种全局检查从一开始就走偏了。

正确的实现方式 #

既然你想用metadata来动态触发数据质量检查，正确的做法是遍历metadata的每一行，针对目标df执行对应的检查，然后把检查结果和metadata合并。示例代码如下：

from pyspark.sql import functions as F

# 先定义你的Valid_df类（假设已经实现了核心检查逻辑）
class Valid_df:
    def __init__(self, df):
        self.df = df
    
    def is_not_null(self, col):
        # 示例实现：返回（检查通过数量，失败数量，是否通过）
        total = self.df.count()
        null_count = self.df.filter(F.col(col).isNull()).count()
        self.result = (total - null_count, null_count, null_count == 0)
        return self
    
    def is_unique(self, col):
        total = self.df.count()
        distinct_count = self.df.select(col).distinct().count()
        self.result = (distinct_count, total - distinct_count, distinct_count == total)
        return self
    
    def execute(self):
        return self.result

# 目标df和metadata定义保持不变
df = spark.createDataFrame( 
    [ (None, 128.0, 1),(110, 127.0, 2),(111, 127.0, 3),(111, 127.0, 4) ,(111, 126.0, 5),(111, 127.0, 6),(109, 126.0, 7),(111, 126.0, 1001) ,(114, 126.0, 1003),(115, 83.0, 1064),(116, 127.0, 1066) ], 
    ['HR', 'maxABP', 'Second'] 
)

metadata = spark.sql("select 'HR' as col, 'is_null' as dq_check")

# 遍历metadata的每一行，执行检查并收集结果
results = []
for row in metadata.collect():
    col_name = row["col"]
    dq_test = row["dq_check"]
    
    if dq_test == 'is_null':
        pass_count, fail_count, is_valid = Valid_df(df).is_not_null(col_name).execute()
    elif dq_test == 'unique':
        pass_count, fail_count, is_valid = Valid_df(df).is_unique(col_name).execute()
    else:
        pass_count, fail_count, is_valid = (0, 0, False)
    
    results.append((col_name, dq_test, pass_count, fail_count, is_valid))

# 把结果转成DataFrame展示
result_df = spark.createDataFrame(results, ["col", "dq_check", "pass_count", "fail_count", "is_valid"])
result_df.show()

补充说明 #

如果你的数据量很大，collect() metadata可能会有性能顾虑，但metadata本身是配置类的小数据集，所以这种方式完全可行。如果要更贴合Spark分布式特性，也可以用广播变量传递目标列的统计信息，但对于列级检查来说，直接遍历metadata的方式更简单直观。

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