PAYATHON 2026

在 apache/airflow 镜像中运行 @task.docker 为何出现 OSError？

数据小陈

问题出在容器内生成脚本的 Airflow supervisor 通信初始化阶段。脚本调用 reinit_supervisor_comms() 时，将一个并非 socket 的文件描述符传给 socket.socket(fileno=fd)，抛出了 OSError: [Errno 88]。对于只执行 print("Foo") 的任务，优先改用不含 Airflow 的 Python 镜像。

为什么安装 Airflow 后才失败？

日志显示，Docker 容器已经创建、启动，并开始执行 /tmp/script.py。异常发生在生成脚本调用 reinit_supervisor_comms() 时，还没有执行到 foo()。

@task.docker 会提取函数源码并生成独立脚本，再将脚本和参数传入容器执行。相关实现复用了 Python virtualenv 脚本生成工具。Docker 装饰器源码

在包含相关初始化逻辑的脚本模板中，如果能够导入 airflow.sdk.execution_time.task_runner，且其中存在 reinit_supervisor_comms，脚本就会调用它。没有安装 Airflow SDK 的容器会跳过这一分支，与问题中的对比现象吻合。脚本模板源码

Airflow 3.2.0 的对应实现通过 __AIRFLOW_SUPERVISOR_FD 获取文件描述符；变量不存在时默认使用 0，然后尝试将其包装成 socket。独立 Docker 容器不会因为安装了 Airflow 就自动获得外层任务进程的 supervisor socket。即使传入同一个描述符数字，也不代表容器内存在对应的 socket。Airflow 3.2.0 Task SDK 源码

日志可以确认该描述符不是 socket，但没有给出它的实际值，因此不能断定此次一定使用了默认值 0。

方案一：使用不含 Airflow 的任务镜像

这个函数只需要 Python，无需在任务容器中安装 apache-airflow。可以保留外层 Airflow 环境，调整 @task.docker 使用的镜像：

from airflow.sdk import dag, task


@task.docker(
    image="python:3.11-slim",
    expect_airflow=False,
)
def foo():
    print("Foo")


@dag(dag_id="minwe_dag")
def minwe_dag():
    foo()


minwe_dag()

如果有业务依赖，可以基于 Python 镜像构建任务镜像，只安装函数实际需要的包。上例补齐了原始粘贴代码中的缩进；日志已经表明实际 DAG 成功解析，当前异常应从容器脚本初始化入手排查。

这个方案依赖任务镜像不再包含会触发初始化分支的 Airflow SDK。expect_airflow=False 表明任务不依赖 Airflow，但不能单独视为这个异常的通用修复。

为什么不能只设置 expect_airflow=False？

这要看执行端实际使用的脚本模板。在已核对的模板中，expect_airflow 控制宏插件加载，而导入和调用 reinit_supervisor_comms() 的逻辑在该条件块之外。因此，如果镜像中仍安装了 Airflow SDK，仅设置 expect_airflow=False 可能继续报错。脚本模板中的条件范围

问题没有提供执行端各 provider 的版本，不能仅凭容器标签确定所用模板，也不能据此指定某个升级版本一定能够修复。

方案二：保留镜像，直接运行脚本

如果必须使用这个 Airflow 镜像，而需求只是执行普通 Python 命令，可以改用 DockerOperator，绕过 @task.docker 自动生成的函数包装脚本：

from airflow.sdk import dag
from airflow.providers.docker.operators.docker import DockerOperator


@dag(dag_id="minwe_dag")
def minwe_dag():
    DockerOperator(
        task_id="foo",
        image="docker.1ms.run/apache/airflow:3.2.0-python3.11",
        entrypoint=["python3"],
        command=["-c", "print('Foo')"],
    )


minwe_dag()

显式设置 entrypoint，可以让这个简单示例直接启动 Python。该方式适用于命令或脚本任务；如果还需要函数返回值自动传回等 TaskFlow 行为，需要另外设计结果传递方式。

如何确认处理有效？

先在执行任务的 Airflow 环境中记录相关版本：

python -m pip show apache-airflow apache-airflow-task-sdk apache-airflow-providers-docker apache-airflow-providers-standard apache-airflow-providers-common-compat

再运行最小 DAG，检查是否输出 Foo、容器是否以退出码 0 结束，以及任务是否成功。如果保留 @task.docker 和 Airflow SDK 后仍然失败，应检查实际生成的 /tmp/script.py 和执行端使用的模板，确认 supervisor 初始化分支是否仍被执行。

日志中的 pandas 警告和最终的 DockerContainerFailedException 不应作为首要修复点：前者没有中断任务，后者只是容器退出码为 1 后的外层异常。直接失败的位置已由 socket.socket(fileno=fd) 的 traceback 给出。

备注：内容仅供参考。