如何构建集成Python图像处理脚本的Airflow DAG流水线

对接你现有Python图像处理脚本到Airflow有两种常用实现路径，可以根据实际运维需求选择：

方案1：最小改动快速对接 #

适合脚本逻辑稳定、需要快速上线调度的场景，几乎不需要修改原有核心处理逻辑，只需要把原本通过命令行传入的参数改为从Airflow配置读取即可。

• 第一步：整理脚本存放路径
  把你的图像处理脚本放到Airflow的DAG加载目录下的可导入模块路径，比如在dags文件夹下新建image_process目录，把脚本（比如命名为img_proc.py）放入该目录，同时新建空文件__init__.py，保证Python可以正常导入模块内的函数。
• 第二步：封装入口逻辑
  不要直接调用脚本里if __name__ == '__main__'块的内容，把这部分初始化逻辑抽成Airflow任务可调用的函数，原有功能函数（process_images/save_processed_data等）完全不需要修改。
• 第三步：编写DAG配置调度规则

示例DAG代码如下：

import os
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
# 直接导入原有脚本里的所有功能函数
from image_process.img_proc import (
    get_df_labels, get_image_ids, process_images, save_processed_data
)

def run_image_pipeline(**context):
    # 配置参数优先从DAG运行参数读取，也可以配置为Airflow Variable统一管理
    config = context['params']
    data_dir = config['data_dir']
    height = config['height']
    width = config['width']
    n_aug = config['n_aug']
    proc_lim = config.get('proc_lim')

    # 以下目录初始化逻辑完全复用原有脚本代码，无需修改
    images_outdir = os.path.join(data_dir, 'processed_images')
    os.makedirs(os.path.dirname(images_outdir), exist_ok=True)

    labels_outdir =  os.path.join(data_dir, 'labels')
    os.makedirs(os.path.dirname(labels_outdir), exist_ok=True)

    ids_outdir = os.path.join(data_dir, 'metadata')
    os.makedirs(os.path.dirname(ids_outdir), exist_ok=True)

    images_outpath_prefix = os.path.join(images_outdir, 'X')
    labels_outpath_prefix = os.path.join(labels_outdir, 'processed')
    ids_outpath_prefix = os.path.join(ids_outdir, 'processed_ids')

    df_labels = get_df_labels()
    image_ids = get_image_ids(df_labels)

    if proc_lim is not None:
        image_ids = image_ids[:proc_lim]
    
    # 执行原有核心处理逻辑
    processed_data = process_images(image_ids)
    save_processed_data(processed_data)

# DAG基础配置
default_args = {
    'owner': 'image-team',
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
}

with DAG(
    dag_id='image_processing_pipeline',
    default_args=default_args,
    schedule='0 2 * * *', # 示例：每天凌晨2点定时执行
    params={
        # 默认运行参数，手动触发DAG时可按需修改
        'data_dir': '/data/image_storage',
        'height': 640,
        'width': 640,
        'n_aug': 3,
        'proc_lim': None
    },
    catchup=False
) as dag:
    full_process_task = PythonOperator(
        task_id='run_full_image_process',
        python_callable=run_image_pipeline,
        execution_timeout=timedelta(hours=4), # 根据实际处理时长调整超时阈值
    )

这个方案的注意事项：

• 提前在所有Airflow Worker节点安装脚本依赖的图像处理库（Pillow、OpenCV、pandas等），否则任务会触发导入错误
• 所有文件路径尽量用绝对路径，避免Airflow运行时工作目录和本地调试目录不一致导致的文件找不到问题
• 如果单批次处理数据量很大，记得调大任务的execution_timeout参数，避免任务跑一半被Airflow强制杀掉

方案2：拆分细粒度任务的生产级DAG #

适合处理数据量大、流程耗时长、需要精细化运维的场景。把原本串行的全流程拆分为独立任务，出错时只需要重跑失败步骤，不用从头跑全量数据，同时可以在Airflow UI直观看到每个步骤的运行状态。
拆分后的任务链路逻辑：

• 初始化任务：完成输出目录创建、标签数据读取、待处理image_ids列表生成，把image_ids存到临时文件，仅将临时文件路径通过XCom传递给下游（注意：不要直接把全量image_ids列表通过XCom传递，数据量大时会拖慢Airflow元数据库）
• 图片处理任务：读取上游传递的临时文件路径，加载image_ids列表执行process_images逻辑，处理完成后把结果临时存储路径传给下游
• 保存校验任务：读取上游的处理结果，执行save_processed_data逻辑，最后增加一步输出文件完整性校验，确认所有结果都写入成功后标记任务完成

拆分后的任务依赖示例：

with DAG(
    dag_id='image_processing_pipeline_split',
    default_args=default_args,
    schedule='0 2 * * *',
    params=default_config,
    catchup=False
) as dag:
    init_task = PythonOperator(
        task_id='init_env_and_load_ids',
        python_callable=init_env_and_get_ids, # 封装目录创建、读标签、生成id列表逻辑
    )

    process_task = PythonOperator(
        task_id='batch_process_images',
        python_callable=run_batch_process, # 封装核心图片处理逻辑
        execution_timeout=timedelta(hours=4),
        pool='image_process_cpu_pool' # 可以单独给计算密集型任务配置资源池
    )

    save_task = PythonOperator(
        task_id='save_and_validate_result',
        python_callable=save_and_check_result, # 封装结果存储、完整性校验逻辑
    )

    init_task >> process_task >> save_task

这个方案的优势：

• 故障恢复成本低：比如保存结果时遇到磁盘满、权限不足的问题，修复后只需要重跑最后一个保存任务，不需要重新跑几小时的图片处理逻辑
• 资源配置灵活：可以给计算密集的图片处理任务单独配置高CPU资源池，轻量的初始化、保存任务用默认资源即可
• 排障效率高：每个任务的日志独立存储，UI上可以直接看到每个步骤的运行时长、失败原因，不用翻全流程日志找报错点

不推荐用BashOperator直接执行python img_proc.py的方式调度，这种方式传参麻烦、任务状态捕获不精准、日志和Airflow集成度差，后续维护成本很高。

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