如何通过Airflow在现有Docker容器上执行Bash命令

你的场景不需要新起容器，直接利用Docker原生的exec能力在已有运行中容器内执行命令即可，两种可落地的实现方式如下：

前置准备 #

不管用哪种方式，先完成两个基础配置：

1. 启动Airflow容器时，将宿主机Docker套接字挂载到Airflow容器内，让Airflow能和宿主机Docker daemon通信：

# 启动Airflow容器时追加挂载参数
-v /var/run/docker.sock:/var/run/docker.sock

2. 给Airflow运行环境配齐操作Docker的依赖：

• 用BashOperator方案：在自定义Airflow镜像中安装docker CLI（Debian系基础镜像执行apt update && apt install -y docker.io，Alpine系执行apk add --no-cache docker）
• 用PythonOperator+DockerHook方案：安装apache-airflow-providers-docker提供商包，版本和你的Airflow版本匹配即可

方案1：BashOperator直接调用docker exec（实现最简单） #

BashOperator本身确实是执行算子运行环境（即Airflow worker容器）本地的命令——只要Airflow worker容器内装了docker CLI，本地执行的docker exec命令本身就可以操作同daemon下的其他运行中容器，不需要新建容器。
示例DAG代码：

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

with DAG(
    dag_id="run_py_in_existing_docker_container",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    run_target_script = BashOperator(
        task_id="exec_py_script",
        # 把<容器名/ID>、<脚本容器内绝对路径>替换成你自己的配置，python3/python根据目标容器内实际命令改
        bash_command="docker exec <你的目标容器名或容器ID> python3 /opt/app/your_target_script.py"
    )

方案2：PythonOperator封装DockerHook调用（日志处理、参数传递更灵活） #

如果不想硬写Shell命令，需要更灵活的参数控制、日志实时采集，可以直接用Airflow官方Docker提供商的Hook封装执行逻辑，本质还是调用Docker API实现exec操作，不会新建容器。
示例DAG代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.docker.hooks.docker import DockerHook
from datetime import datetime

def run_script_in_container():
    # 初始化Docker Hook，提前在Airflow连接页配置docker_default连接，Host填unix://var/run/docker.sock
    docker_hook = DockerHook(
        docker_conn_id="docker_default",
        version="auto",
        tls=False
    )
    # 获取已在运行的目标容器
    target_container = docker_hook.client.containers.get("<你的目标容器名或容器ID>")
    # 执行容器内命令，开启流日志输出
    exec_result = target_container.exec_run(
        cmd=["python3", "/opt/app/your_target_script.py"],
        stream=True,
        demux=True
    )
    exit_code, output_stream = exec_result
    # 实时打印stdout/stderr日志
    for stdout_line, stderr_line in output_stream:
        if stdout_line:
            print(stdout_line.decode().strip())
        if stderr_line:
            print(stderr_line.decode().strip())
    # 非0退出码抛错让任务标记失败
    if exit_code != 0:
        raise RuntimeError(f"容器内脚本执行异常，退出码：{exit_code}")

with DAG(
    dag_id="run_py_in_existing_container_via_hook",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    run_target_script = PythonOperator(
        task_id="exec_py_script",
        python_callable=run_script_in_container
    )

注意事项 #

• 所有命令里的脚本路径是 目标容器内的绝对路径，不是Airflow容器路径，也不是宿主机路径
• 如果执行时提示docker.sock权限不足，把Airflow进程运行用户加入宿主机docker用户组即可，生产环境不建议直接给sock文件设置777权限
• 目标容器必须处于running状态，否则exec命令会执行失败，可以在任务前加个传感器判断容器状态
• 原生DockerOperator的设计逻辑就是每次新建独立容器执行任务，执行完成后默认销毁容器，确实不满足操作已有容器的需求

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