Prefect运行流报Task slug不存在KeyError 本地正常Web UI Quick Run失败

核心报错原因 #

• Prefect的任务运行时存在上下文绑定机制，本地直接调用flow.run()时每次运行都会生成独立的上下文，所以不会出现冲突；但通过Web UI运行注册后的Flow时，所有任务的元数据会在注册阶段完成序列化，你在loop_cheking_files_exists任务内部定义子Flow，子Flow中用到的全局Prefect任务会默认绑定到主Flow的上下文，子Flow自身没有对应任务的元数据，运行时查找任务slug就会抛出KeyError。
• 你在子Flow定义中直接调用任务的.run()方法属于本地调试用法，注册到远端后这种调用会跳过Prefect的元数据注册逻辑，进一步加剧上下文冲突。
• 单个任务内部直接定义并运行子Flow不符合Prefect 1.x的流编排规范，任务和Flow的作用域没有隔离，会导致调度逻辑混乱。

修复方案 #

步骤1：调整子Flow的作用域 #

将子Flow从任务内部移到全局作用域定义，子Flow的输入通过Parameter传递，示例如下：

# 全局作用域定义子Flow
with Flow('subflow', result=PrefectResult()) as subflow:
    # 子Flow的输入通过Parameter声明
    run_date = Parameter("run_date")
    date_diff, last_file_date = date_difference(run_date)
    
    with case(date_diff, True):
        new_date = last_file_date + timedelta(days = 1)
        cond_log_exist = check_file_exists(path_ftp, log_file_name)

        with case(cond_log_exist, True):
            false_result = cond_log_exist
            row_count, md5_sum = check_log_result(path_ftp, log_file_name)
            merging = merge(row_count, md5_sum)
            download = file_download(zip_file_name, path_ftp, path_interim, upstream_tasks=[merging])
            unzip = unzip_sku_file(path_interim, zip_file_name, upstream_tasks=[download])
            check_skunames_file = check_row_md5(path_interim, csv_file_name, row_count, md5_sum, upstream_tasks=[unzip])
        
            with case(check_skunames_file, True):
                upload = file_download(zip_file_name, path_interim, path_pri+f'\{new_date_yyyy}010{new_date_mm}', remove=True)
                msg_delete_csv_file = delete_file(csv_file_name, path_interim, upstream_tasks=[upload])
                write_zipname = write_zip_name(zip_file_name, upstream_tasks=[msg_delete_csv_file])

            with case(check_skunames_file, False):
                print('check_skunames_file, False')
                fail_check_skunames = fail_email_notification()

            new_date_diff, last_file_date = date_difference(run_date, upstream_tasks=[write_zipname])
            success_notification = success_email_notification(zip_file_name, upstream_tasks=[write_zipname])

        with case(cond_log_exist, False):
            fail_notification = fail_email_notification(log_file_name, upstream_tasks=[cond_log_exist])
            false_result = action_if_false(upstream_tasks=[fail_notification])

步骤2：修改循环任务的子Flow调用逻辑 #

使用FlowRunner运行子Flow，完全隔离主/子Flow的上下文，修改loop_cheking_files_exists任务如下：

from prefect.engine.flow_runner import FlowRunner

@task(name='loop_checking_files_exist', result=PrefectResult())
def loop_cheking_files_exists(run_date):
    # 用FlowRunner运行子Flow，避免上下文冲突
    runner = FlowRunner(flow=subflow)
    subflow_run_id = runner.run(parameters={"run_date": run_date})
    # 从子Flow实例中获取对应任务的结果，避免引用主Flow的任务对象
    date_diff_task = subflow.get_tasks(name="date_difference")[-1]
    child_data = subflow_run_id.result[date_diff_task]._result.value

    if child_data == True:
        raise LOOP(message=f'Child data = {child_data}')
    else: 
        raise FAIL()

步骤3：修正主Flow的时间参数 #

不要直接用datetime.now()，该时间会在Flow注册时固定，不会随运行时间更新，替换为Prefect内置的时间工具：

from prefect.utilities.dates import utcnow

with Flow('Main Flow Name', result=PrefectResult()) as flow:
    date = utcnow()
    loop_task = loop_cheking_files_exists(date)

    with case(loop_task, True):
        success_notification = success_email_notification()

    with case(loop_task, False):
        fail_notification = fail_email_notification()

步骤4：调整Flow注册配置 #

如果使用Local存储，注册时开启stored_as_script=True，确保子Flow的定义会被完整加载：

if __name__ == "__main__":
    flow.storage = Local(path="你的脚本完整路径", stored_as_script=True)
    flow.register('Project_Name')

内容的提问来源于stack exchange，提问作者S.Honcharov