PySpark数据集行存在性检查代码报TypeError错误求助

错误原因 #

你的代码中@transform_df装饰器仅声明了violations作为输入参数，但compute函数却要求传入output参数。在Transforms API中，若要在函数内访问输出数据集，必须在@transform_df里显式将Output对象以命名参数的形式定义，否则函数无法获取到output的引用。

另外你的filter_data函数定义了output参数但实际并未使用，属于冗余参数，可直接删除。

修正后的代码 #

from transforms.api import configure, transform_df, Input, Output
from numpy import mean, std
from datetime import timedelta, datetime
from pyspark.sql import functions as F

@configure(profile=[
    "DRIVER_MEMORY_MEDIUM"
    ])
@incremental(snapshot_inputs=['violations'])
@transform_df(
    output=Output("/Spring/Snowplow/data/derived/output_dataset"),
    violations=Input("xxx"),
)
def compute(violations, output):
    # 调用filter_data时不再传入output参数，因为函数内部不需要它
    alerts = filter_data(violations)
    # 计算alerts中存在但output中没有的行
    new_alerts = alerts.subtract(output)
    
    if new_alerts.count() > 0:
        return alerts
    else:
        return output

def filter_data(violations):
    todays_date = F.current_date()
    violations = violations.filter(F.col("failure_date") == todays_date)

    columns_to_keep = [
        "id",
        "value_1",
        "value_2",
        "date",
        "timestamp",
    ]
    return violations.select(*columns_to_keep)

额外优化建议 #

使用count()判断是否有新行在数据量较大时效率较低，可改用左反连接的方式快速判断是否存在未匹配的行，提升性能：

# 替换原有的new_alerts.count()判断逻辑
has_new_rows = alerts.join(output, on="id", how="left_anti").limit(1).count() > 0
if has_new_rows:
    return alerts
else:
    return output

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