Airflow调用Docker中其他容器PHP任务报错求助

1. 搞定ModuleNotFoundError: No module named 'command' #

Python根本没有叫command的标准库，PyPI上也找不到这个包，所以你跑pip install command纯粹是白忙活。得换用正确的模块：

• 如果是要执行系统命令，用Python自带的subprocess模块就行，不用额外安装；
• 如果想直接操作Docker容器，就装docker SDK（执行pip install docker）。

2. 解决Object of type bytes is not JSON serializable #

这个报错是因为你执行命令后拿到的stdout/stderr是字节类型（bytes），而Airflow在记录日志或者传递XCom的时候，要求数据是可JSON序列化的类型（比如字符串）。解决办法很简单：

• 用subprocess.check_output(..., text=True)参数直接让输出返回字符串；
• 或者手动把字节流解码成字符串：output.decode('utf-8')。

3. 解决Docker API权限拒绝问题PermissionError(13, 'Permission denied') #

这个问题是Airflow容器里的用户没权限访问宿主机的Docker socket（/var/run/docker.sock），按下面步骤改：

1. 在Airflow的docker-compose.yml里，给webserver和worker服务挂载Docker socket：

   volumes:
     - /var/run/docker.sock:/var/run/docker.sock
   

2. 让容器内的用户加入有权限访问socket的用户组：
   • 先在宿主机跑ls -l /var/run/docker.sock，看输出里的组ID（通常是999，对应docker组）；
   • 在Airflow服务的配置里指定用户组：

     user: "${AIRFLOW_UID:-50000}:999"
     

   把999换成你刚才查到的实际组ID就行。

两种可行的实现方案 #

方案一：用subprocess执行Docker命令

修正后的DAG代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import subprocess

def run_php_task():
    # 替换成你的PHP容器名称和脚本路径
    cmd = [
        'docker', 'exec', 'your-php-container-name',
        'php', '/path/to/your/task.php'
    ]
    try:
        # text=True 直接返回字符串，避免bytes序列化问题
        result = subprocess.check_output(cmd, text=True, stderr=subprocess.STDOUT)
        print(f"PHP任务执行结果: {result}")
        return result
    except subprocess.CalledProcessError as e:
        print(f"PHP任务执行失败: {e.output}")
        raise e  # 抛出异常让Airflow标记任务失败

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'run_php_docker_task',
    default_args=default_args,
    description='通过Airflow调用Docker内的PHP任务',
    schedule_interval=timedelta(days=1),
    catchup=False,
) as dag:

    run_php = PythonOperator(
        task_id='execute_php_task',
        python_callable=run_php_task,
    )

    run_php

方案二：用Docker SDK直接操作容器

1. 先给Airflow容器装Docker SDK：可以在docker-compose.yml的服务里加command先安装，或者进入容器跑pip install docker；
2. 修正后的DAG代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import docker

def run_php_task():
    # 初始化Docker客户端
    client = docker.from_env()
    # 替换成你的PHP容器名称和脚本路径
    exec_result = client.containers.get('your-php-container-name').exec_run(
        cmd=['php', '/path/to/your/task.php'],
        stdout=True,
        stderr=True
    )
    # 把字节输出转成字符串
    stdout = exec_result.output.decode('utf-8')
    if exec_result.exit_code != 0:
        print(f"PHP任务执行失败: {stdout}")
        raise Exception(f"任务失败，错误信息: {stdout}")
    print(f"PHP任务执行成功: {stdout}")
    return stdout

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'run_php_docker_task',
    default_args=default_args,
    description='通过Airflow调用Docker内的PHP任务',
    schedule_interval=timedelta(days=1),
    catchup=False,
) as dag:

    run_php = PythonOperator(
        task_id='execute_php_task',
        python_callable=run_php_task,
    )

    run_php

最后调整Airflow的docker-compose.yml #

确保webserver和worker服务添加以下配置：

services:
  airflow-webserver:
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock  # 挂载Docker socket
    user: "${AIRFLOW_UID:-50000}:999"  # 替换999为你宿主机docker组的实际ID
    # 保留原有其他配置...

  airflow-worker:
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    user: "${AIRFLOW_UID:-50000}:999"
    # 保留原有其他配置...

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