Airflow TaskFlow API中FileSensor与@task依赖配置报错求助

错误原因 #

你遇到的AttributeError: 'function' object has no attribute 'update_relative'是因为：

• 被@task装饰的download本质还是Python函数，并非Airflow的Task实例
• >>依赖运算符只能在Airflow Task对象之间生效，直接传递函数会导致Airflow无法识别并设置任务依赖

修复方案 #

需要先通过调用@task装饰的函数生成Task实例，再与FileSensor建立依赖，同时避免重复调用函数生成多个独立Task：

修正后的代码示例 #

@dag(...)  # 补充你的DAG参数，比如schedule_interval、start_date等
def my_dag():

    if_file_exists = FileSensor(
            task_id="file_sensor",
            filepath="your_file_path", 
            fs_conn_id="your_fs_conn",
            poke_interval=30,
        )

    # 调用@task装饰的函数生成Task实例
    download_task = download()

    # 建立FileSensor与download任务的依赖
    if_file_exists >> download_task

    # 后续任务复用同一个Task实例，避免重复创建任务
    show(transform(download_task)) >> upload()

关键说明 #

• 每个@task函数调用都会生成一个独立的Task对象，务必将调用结果赋值给变量，确保上下游依赖的是同一个Task实例
• 传统Operators（如FileSensor）和TaskFlow任务配置依赖的核心是：必须基于Airflow的Task对象而非原始Python函数

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