如何捕获容器退出码并用于Airflow任务的状态判定

具体实现方案 #

方案1：直接使用DockerOperator #

Airflow的DockerOperator默认会根据容器退出码判定任务状态：退出码0标记成功，非0标记失败，完全匹配你的需求。示例代码如下：

from airflow import DAG
from airflow.providers.docker.operators.docker import DockerOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
}

with DAG('container_exit_code_dag', default_args=default_args, schedule_interval='@daily') as dag:
    # 第一个任务示例
    task1 = DockerOperator(
        task_id='first_task',
        image='your-first-image:latest',
        command='echo "First task finished"',
        docker_url='unix://var/run/docker.sock',
        network_mode='bridge'
    )

    # 第二个任务：自动根据容器退出码0成功、1失败
    task2 = DockerOperator(
        task_id='second_task',
        image='your-second-image:latest',
        command='your-container-exec-command',  # 替换为你的容器执行命令
        docker_url='unix://var/run/docker.sock',
        network_mode='bridge'
    )

    task1 >> task2

方案2：用BashOperator自定义退出码逻辑 #

如果你的容器可能返回其他非0退出码，且希望仅在退出码为1时标记失败，其他非0码按成功处理，可通过Bash脚本捕获并判断：

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
}

with DAG('custom_exit_code_dag', default_args=default_args, schedule_interval='@daily') as dag:
    task1 = BashOperator(
        task_id='first_task',
        bash_command='echo "First task completed"'
    )

    task2 = BashOperator(
        task_id='second_task',
        bash_command='''
            # 运行容器并捕获退出码
            docker run --rm your-second-image:latest your-container-command
            EXIT_CODE=$?
            # 自定义逻辑：0成功，1失败，其他退出码按成功处理
            if [ $EXIT_CODE -eq 0 ]; then
                exit 0
            elif [ $EXIT_CODE -eq 1 ]; then
                exit 1
            else
                exit 0
            fi
        '''
    )

    task1 >> task2

关键注意事项 #

• 确保Airflow Worker有权限访问Docker守护进程（比如将Worker用户加入docker组，或配置正确的docker_url）。
• 自定义镜像需确保Airflow环境可访问（本地存在或已推送到镜像仓库）。
• 如需更复杂的状态判定，可结合PythonOperator调用Docker SDK获取退出码，再自定义任务状态。

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