如何让Airflow任务循环中失败的DatabricksRunNowOperator不阻塞后续任务

我有一个Airflow DAG，需根据用户在params中提供的特定时段运行外部作业，流程如下：

• 列出指定日期范围内S3存储桶中的文件，验证特定文件存在并返回映射{partition -> {start_date, end_date}}。
• 利用第一步返回的映射，按分区动态创建多个"run_job"任务并并行运行。
• 在每个"run_job"中遍历时段列表{start_date, end_date}，逐个顺序启动新的DatabricksRunNowOperator执行。

以下是DAG的简化版本：

with DAG(
        dag_id="backfill",
        start_date=datetime(2023, 6, 1),
        catchup=False,
        render_template_as_native_obj=True,
        max_active_tasks=10,
        schedule=None,
        params={
            "start_date": Param(DEFAULT_DATE_PARAM, format="date", type="string"),
            "end_date": Param(DEFAULT_DATE_PARAM, format="date", type="string"),
        },
) as backfill:
    @task
    def run_job(period_to_partition: (str, [str, str]), **context):
        tasks = []
        partition, periods = period_to_partition

        for period in periods:
            start_date = period.get("start_date")
            end_date = period.get("end_date")

            job_params = prepare_job_args(start_date=start_date, end_date=end_date)

            run_job = DatabricksRunNowOperator(
                job_id="job_id",
                task_id=f"run_dbx_job_{partition}_{start_date}_{end_date}",
                spark_submit_params=job_params,
                dag=backfill,
                trigger_rule=TriggerRule.ALWAYS,
                retries=3
            )
            tasks.append(run_job)

        chain(*tasks)

        for single_task in tasks:
            single_task.execute(context)


    @task
    def prepare_period(**context):
        """List files in S3 bucket for the date range, verify that specific files are present 
        and return a map {partition -> {start_date, end_date}}."""
        params = context["params"]
        return split_to_partition_to_periods(params["start_date"], params["end_date"], context)


    run_job.expand(period_to_partition=prepare_period())

上述代码正常运行时没问题，但如果某个DatabricksRunNowOperator失败，tasks数组中的后续任务会直接终止，整个DAG标记为失败。我试过几种方案都没解决：

• 移除链式调用并给算子加trigger_rule=ALL_DONE，但参数似乎被外层任务覆盖，无效。
• 无法在任务外查询S3，因为需要用户传入的start_date和end_date参数。
• 无法把算子创建逻辑移到其他任务或DAG级别，因为依赖任务返回的时段信息才能确定算子数量。
• DAG级默认参数只对外部任务生效，内部算子不继承。

核心疑问：有没有办法让单个DatabricksRunNowOperator失败后，其余算子仍能继续执行？

你的问题根源是在单个Airflow Task内部直接执行Operator实例，这种方式把多个Databricks作业绑定到同一个Task中，一旦某个作业失败，整个Task会终止，后续作业自然无法运行。Airflow的Operator设计是作为独立Task在DAG任务图中运行，而非嵌套在另一个Task内部执行。以下是两种可行的解决方法：

方法1：嵌套动态任务映射（Airflow 2.3+ 支持） #

利用Airflow的嵌套动态映射能力，让每个Databricks作业成为独立Task，单个作业失败不会影响其他作业执行，还能在UI中单独查看、重试每个作业。

改造步骤： #

1. 修改prepare_period任务，返回扁平化的参数列表[(partition, start_date, end_date), ...]，替代原有的分区分组结构。
2. 用DatabricksRunNowOperator.partial结合动态映射，生成独立的作业Task；如果需要按分区顺序执行时段任务，可先分组再做二次映射。

示例代码：

from airflow.decorators import task
from airflow.operators.databricks import DatabricksRunNowOperator
from airflow.models.param import Param
from datetime import datetime
from airflow.utils.trigger_rule import TriggerRule
from airflow.models.baseoperator import chain

def prepare_job_args(start_date, end_date):
    # 保留你的参数准备逻辑
    return {"--start-date": start_date, "--end-date": end_date}

def split_to_partition_to_periods(start_date, end_date, context):
    # 修改为返回扁平化列表，替换为你的真实逻辑
    return [
        ("partition1", "2023-06-01", "2023-06-02"),
        ("partition1", "2023-06-03", "2023-06-04"),
        ("partition2", "2023-06-01", "2023-06-02")
    ]

DEFAULT_DATE_PARAM = "2023-06-01"

with DAG(
        dag_id="backfill",
        start_date=datetime(2023, 6, 1),
        catchup=False,
        render_template_as_native_obj=True,
        max_active_tasks=10,
        schedule=None,
        params={
            "start_date": Param(DEFAULT_DATE_PARAM, format="date", type="string"),
            "end_date": Param(DEFAULT_DATE_PARAM, format="date", type="string"),
        },
) as backfill:
    @task
    def prepare_period(**context):
        params = context["params"]
        return split_to_partition_to_periods(params["start_date"], params["end_date"], context)

    # 按分区分组，用于后续按分区顺序执行时段任务
    @task
    def group_by_partition(flat_items):
        partition_groups = {}
        for item in flat_items:
            partition = item[0]
            if partition not in partition_groups:
                partition_groups[partition] = []
            partition_groups[partition].append((item[1], item[2]))
        return [(k, v) for k, v in partition_groups.items()]

    grouped_periods = group_by_partition(prepare_period())

    # 单个分区内的时段任务逻辑
    @task
    def run_partition_jobs(partition_periods):
        partition, periods = partition_periods
        tasks = []
        for start_date, end_date in periods:
            task = DatabricksRunNowOperator(
                task_id=f"run_dbx_job_{partition}_{start_date}_{end_date}",
                job_id="job_id",
                spark_submit_params=prepare_job_args(start_date, end_date),
                retries=3,
                trigger_rule=TriggerRule.ALL_DONE  # 确保前面任务失败后，后续任务仍执行
            )
            tasks.append(task)
        chain(*tasks)  # 保持时段任务顺序执行
        return tasks

    # 动态映射每个分区的任务组
    run_partition_jobs.expand(partition_periods=grouped_periods)

方法2：Task内部捕获异常继续执行 #

如果你的Airflow版本低于2.3，无法使用嵌套动态映射，可以在run_job任务内部捕获每个DatabricksRunNowOperator.execute()的异常，记录错误后继续执行后续作业。

示例代码：

@task(trigger_rule=TriggerRule.ALL_DONE)
def run_job(period_to_partition: (str, [str, str]), **context):
    partition, periods = period_to_partition
    failed_jobs = []

    for period in periods:
        start_date = period.get("start_date")
        end_date = period.get("end_date")
        job_params = prepare_job_args(start_date=start_date, end_date=end_date)

        run_job_op = DatabricksRunNowOperator(
            job_id="job_id",
            task_id=f"run_dbx_job_{partition}_{start_date}_{end_date}",
            spark_submit_params=job_params,
            dag=backfill,
            retries=3
        )

        try:
            run_job_op.execute(context)
        except Exception as e:
            # 记录失败作业信息
            failed_jobs.append(f"{partition}_{start_date}_{end_date}: {str(e)}")
            # 跳过当前失败任务，继续执行后续任务
            continue

    # 可选：如果有失败任务，抛出异常让外层任务标记为失败
    if failed_jobs:
        raise Exception(f"部分Databricks作业失败: {', '.join(failed_jobs)}")

注意事项： #

• 方法1是Airflow推荐的最佳实践，每个Databricks作业都是独立Task，可单独管理、重试，状态更清晰。
• 方法2中，外层run_job任务的状态由你决定：如果最后抛出异常，任务标记为失败；如果不抛出，任务标记为成功，但需自行记录失败信息以便后续排查。

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