Airflow成功/失败回调中Databricks Notebook调用不生效问题

你在成功/失败回调函数中仅完成了DatabricksSubmitRunOperator的对象实例化，并未触发其实际执行逻辑。
Airflow的Operator执行机制分为两个独立环节：

• DAG静态解析阶段：所有被绑定到DAG对象的Operator会被注册到任务依赖图，由调度器统一触发执行
• 任务运行时阶段：回调函数属于任务运行时的执行逻辑，在回调内手动创建的Operator没有被注册到DAG调度链路，仅为存在于内存中的无引用对象，不会被自动调度

你现有代码中创建的DatabricksSubmitRunOperator实例既没有被变量引用，也没有执行任何方法调用，属于完全无效的内存对象，自然无法提交Databricks作业。

方案1：回调内手动触发Operator执行 #

在创建DatabricksSubmitRunOperator实例后，直接调用其execute方法，将当前任务的上下文传入，即可绕开DAG注册流程，直接提交Databricks作业。
修改后的回调代码示例片段：

def task_success_callback(context):
    print("success case")
    dq_notebook_success_task_params = {
        'existing_cluster_id': Variable.get("DATABRICKS_CLUSTER_ID"),
        'notebook_task': {
            'notebook_path': '/AAA/Airflow/Operators/audit_file_operator',
             'base_parameters': {
                "root": "dbfs:/mnt/aaa",
                "audit_file_path": "/success_file_path/",
                "table_name": "sample_data_table",
                "audit_flag": "success"
            }
        }
    }
    # 初始化Operator实例并保存引用
    db_submit_operator = DatabricksSubmitRunOperator(
        task_id=f"success_audit_{context['task_instance'].task_id}",
        databricks_conn_id='databricks_conn',
        json=dq_notebook_success_task_params,
        do_xcom_push=True,
        secrets=[secret.Secret(
            deploy_type='env',
            deploy_target=None,
            secret='adf-service-principal'
        ), secret.Secret(
            deploy_type='env',
            deploy_target=None,
            secret='postgres-credentials',
        )],
    )
    # 缺失的核心逻辑：手动触发Operator执行
    db_submit_operator.execute(context=context)

注意：这种方式提交的Databricks作业不会在Airflow UI中生成独立的任务运行记录，执行日志会合并到触发回调的原任务日志中，不支持独立的重试、状态追踪配置。

方案2：采用原生DAG依赖实现分支触发（推荐） #

脱离回调内写执行逻辑的模式，在DAG构建阶段直接定义和原业务任务绑定的分支依赖，完全符合Airflow的调度设计逻辑：

• 原业务任务执行完成后，通过BranchPythonOperator判断其运行状态（成功/失败）
• 两个判断分支分别挂载对应的Databricks Notebook提交任务

这种模式下所有Databricks作业都是受Airflow原生管理的任务，有独立的运行实例、日志、重试机制，稳定性远高于回调内手动执行的方式。

• 不要在回调内硬编码重复的task_id：手动调用Operator执行时，重复的固定task_id可能导致运行时的元数据冲突，建议结合触发回调的原任务ID动态生成唯一task_id
• 确认运行时环境依赖：回调内手动执行逻辑时，需要确保执行回调的Worker节点可以正常读取到Airflow配置的Databricks连接、Variable变量值，避免出现权限、连接参数读取失败的问题

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