如何通过Docker部署Airflow并使用不同worker实现多Python环境隔离

完全可以通过自定义Worker镜像+队列路由的方式实现不同项目依赖完全隔离，操作步骤如下：

1. 构建项目专属的Worker镜像 #

分别为不同项目构建独立的Airflow Worker镜像：

• 基于官方Airflow Worker基础镜像，安装项目1所需的全部Python依赖，打镜像标签为 airflow-worker-project1:latest
• 同理构建项目2的Worker镜像，安装项目2专属依赖，打镜像标签为 airflow-worker-project2:latest

2. 修改docker-compose配置，启动多队列Worker #

官方docker-compose默认使用CeleryExecutor，你只需要把原有的单Worker服务替换为两个监听不同队列的Worker服务即可，配置示例如下：

# 注释或删除原有默认worker服务，新增以下两个worker服务
  worker-project1:
    image: airflow-worker-project1:latest
    command: celery worker -q project1_queue
    environment:
      <<: *airflow-common-env
    volumes:
      - ./dags:/opt/airflow/dags
      - ./logs:/opt/airflow/logs
      - ./plugins:/opt/airflow/plugins
    restart: always
    depends_on:
      - airflow-webserver
      - airflow-scheduler
      - redis

  worker-project2:
    image: airflow-worker-project2:latest
    command: celery worker -q project2_queue
    environment:
      <<: *airflow-common-env
    volumes:
      - ./dags:/opt/airflow/dags
      - ./logs:/opt/airflow/logs
      - ./plugins:/opt/airflow/plugins
    restart: always
    depends_on:
      - airflow-webserver
      - airflow-scheduler
      - redis

其中-q参数指定该Worker专属监听的队列名称，两个Worker只会处理对应队列的任务。

3. DAG任务指定路由队列 #

编写DAG时为对应项目的任务指定队列参数即可，示例如下：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# 项目1的DAG示例
with DAG(
    dag_id="project1_demo_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False
) as dag:
    project1_task = PythonOperator(
        task_id="run_project1_script",
        python_callable=your_project1_exec_func,
        queue="project1_queue" # 指定该任务路由到项目1的Worker执行
    )

# 项目2的DAG示例
with DAG(
    dag_id="project2_demo_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False
) as dag:
    project2_task = PythonOperator(
        task_id="run_project2_script",
        python_callable=your_project2_exec_func,
        queue="project2_queue" # 指定该任务路由到项目2的Worker执行
    )

所有类型的Airflow Operator都支持queue参数配置，配置后任务会自动下发到对应队列的Worker上运行，两个Worker的Python环境完全独立，不会出现依赖冲突问题。

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