如何在Airflow Kubernetes中运行含独立Dockerfile的新流水线？

核心逻辑 #

Airflow实现自定义镜像流水线的核心是KubernetesPodOperator——让任务（或整条流水线）运行在你基于独立Dockerfile构建的镜像中，和Dagster的模式逻辑一致，只需完成镜像构建、权限配置、DAG编写三个核心环节。

步骤1：构建并推送自定义镜像 #

先编写你的独立Dockerfile，包含流水线所需的依赖、脚本：

# 基于Python基础镜像（也可选用Airflow官方镜像作为基础，按需调整）
FROM python:3.10-slim

# 安装流水线依赖
RUN pip install pandas requests sqlalchemy

# 复制流水线脚本到容器内
COPY my_pipeline_logic.py /app/

# 设置容器工作目录
WORKDIR /app

构建镜像并推送到你的容器镜像仓库（Docker Hub、Harbor、GCR等均可）：

docker build -t your-registry/your-custom-pipeline:v1 .
docker push your-registry/your-custom-pipeline:v1

步骤2：配置Airflow的Kubernetes权限 #

确保Airflow的服务账号拥有在K8s集群中创建Pod的权限：

• 给Airflow服务账号绑定edit角色（或更精细的权限，按需调整）：

kubectl create rolebinding airflow-pod-creator --clusterrole=edit --serviceaccount=airflow:airflow -n airflow

（注：假设Airflow部署在airflow命名空间，服务账号为airflow，请根据实际环境修改）

步骤3：编写DAG调用自定义镜像 #

在Airflow的DAG文件中，用KubernetesPodOperator指定自定义镜像并执行流水线：

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from datetime import datetime

default_args = {
    'owner': 'data-team',
    'start_date': datetime(2024, 1, 1),
    'retries': 1
}

with DAG(
    'custom_image_data_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    catchup=False
) as dag:

    run_pipeline = KubernetesPodOperator(
        task_id='execute_custom_pipeline',
        # 指定你的自定义镜像地址
        image='your-registry/your-custom-pipeline:v1',
        # 镜像拉取策略：Always/IfNotPresent/Never，按需设置
        image_pull_policy='Always',
        # 容器内执行的命令及参数
        cmds=['python'],
        arguments=['/app/my_pipeline_logic.py'],
        # Airflow所在的K8s命名空间
        namespace='airflow',
        # 任务完成后自动删除Pod（可选，默认True）
        is_delete_operator_pod=True,
        # 私有镜像仓库拉取密钥（如果镜像存放在私有仓库）
        image_pull_secrets=['registry-secret'],
        # 资源限制（可选，避免资源不足或浪费）
        resources={
            'request_memory': '2Gi',
            'request_cpu': '1',
            'limit_memory': '4Gi',
            'limit_cpu': '2'
        }
    )

    run_pipeline

关键参数说明 #

• image：必填项，填写你的自定义镜像完整地址
• image_pull_secrets：若镜像在私有仓库，需提前在K8s中创建镜像拉取密钥，此处填写密钥名称
• cmds & arguments：对应容器内要执行的命令和参数，和本地运行python my_pipeline_logic.py的逻辑一致
• resources：按需配置CPU、内存的请求和限制值

进阶：多任务拆分流水线 #

如果流水线有多个步骤，每个步骤可使用独立镜像，或部分步骤用自定义镜像：

# 示例：拆分提取、转换、加载三个步骤
extract_task = KubernetesPodOperator(
    task_id='extract_data',
    image='your-registry/data-extract:v1',
    cmds=['python'],
    arguments=['/app/extract.py']
)

transform_task = KubernetesPodOperator(
    task_id='transform_data',
    image='your-registry/data-transform:v1',
    cmds=['python'],
    arguments=['/app/transform.py']
)

load_task = KubernetesPodOperator(
    task_id='load_data',
    image='your-registry/data-load:v1',
    cmds=['python'],
    arguments=['/app/load.py']
)

extract_task >> transform_task >> load_task

注意事项 #

1. 镜像访问：确保K8s节点能访问你的镜像仓库，私有仓库必须配置image_pull_secrets
2. 日志查看：Airflow会自动收集Pod日志，可直接在Airflow UI的任务详情页查看
3. 依赖完整性：自定义镜像需包含流水线所有依赖，避免运行时缺失
4. 版本管理：镜像必须打版本标签，方便迭代更新和回滚

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