Airflow v1.9 Python Operator执行报错：参数异常求助

这个问题我之前也碰到过，核心原因就是Airflow的上下文参数传递和你的函数定义不兼容，我给你拆解一下问题和解决办法：

错误原因解析 #

你设置了provide_context=True，这时候Airflow会自动把一整套上下文变量（比如next_execution_date、prev_execution_date、dag等十多个参数）传递给你的fill_nulls函数，但你的函数只定义了ds和file_in两个参数，多余的参数就触发了TypeError。

快速解决方案 #

方案1：让函数兼容额外参数（最推荐） #

修改函数定义，添加**kwargs来接收所有Airflow传递的额外参数，这样不管传什么都不会报错：

def fill_nulls(ds, file_in, **kwargs):  # 新增**kwargs接收多余参数
    # 处理glob返回的列表：确保找到文件，否则抛出异常
    csv_files = glob.glob(os.path.join(r'/tmp/', file_in))
    if not csv_files:
        raise FileNotFoundError(f"找不到匹配的文件：/tmp/{file_in}")
    csv_file = csv_files[0]
    
    df = pd.read_csv(csv_file, sep='\t', header=None, error_bad_lines=False, index_col=False, dtype='unicode')
    df = df.fillna(r'\N')
    # 注意：原代码这里没有赋值，替换操作不会生效，必须把结果写回DataFrame
    df.loc[:, df.dtypes == object] = df.loc[:, df.dtypes == object].apply(
        lambda s: s.str.replace(" ", r'\N')
    )
    df.to_csv(csv_file, sep='\t', header=None, index=False, quoting=csv.QUOTE_NONE)

方案2：关闭上下文传递，显式传参 #

如果你不需要Airflow的上下文变量，可以把provide_context=True改成False，然后通过op_kwargs直接传递模板化参数：

# 注意：变量名不要和函数名重复！原代码里变量名和函数名都是fill_nulls，会覆盖函数定义
fill_nulls_task = PythonOperator(
    task_id='fill_nulls',
    python_callable=fill_nulls,
    provide_context=False,
    op_kwargs={
        'ds': '{{ ds }}',
        'file_in': 'apollo_export_{{macros.ds_format(macros.ds_add(ds, -2),\'%Y-%m-%d\',\'%Y%m%d\')}}.csv'
    },
    dag=dag
)

额外的代码优化点 #

除了参数问题，你的代码还有两个容易忽略的bug：

• glob返回列表的处理：glob.glob返回的是匹配到的文件路径列表，如果没有匹配到文件，直接传给pd.read_csv会报错，所以一定要先判断列表是否为空。
• 字符串替换未生效：原代码里的apply操作只是计算了结果，但没有赋值回DataFrame，导致空格替换成\N的操作根本不会生效，必须把结果重新赋值给对应的列。

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