Airflow手动触发任务时ffmpeg进程休眠，任务无法完成求助

安全阿岳

2026-7-18

问题描述

在MacBook Air（M1芯片）上使用Airflow v2.6.1 standalone模式启动，配置为LocalExecutor+PostgreSQL，Python版本3.7.16。通过PythonOperator调用subprocess执行ffmpeg命令时，简单命令（如ls、ffmpeg -version）可成功完成，但执行avi转mp4这类复杂视频处理命令时，UI手动触发后ffmpeg进程处于休眠状态，任务一直无法结束；但使用airflow dags test命令在终端执行该DAG却能快速完成。

相关代码

from __future__ import annotations

import os
from airflow.operators.python import BranchPythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule
from airflow.exceptions import AirflowException
import subprocess
import shutil
input_dir = "/Users/lll/airflow/video_input/"
output_dir = "/Users/lll/airflow/video_output/"

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

with DAG(
    "ffmpeg_example",
    default_args={
        "retries": 2,
    },
    description="cut video into frame and add watermarks",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["ffmpeg"],
) as dag:
    dag.doc_md = __doc__

    # count files
    def count_func():
        file = open("/Users/lll/airflow/debug1.log", "a")
        file.write("start\n")
        file.close()
        file_list = os.listdir(input_dir)
        cnt = 0
        for item in file_list:
            if os.path.isfile(input_dir+item):
                cnt += 1
                os.rename(input_dir+item, input_dir+"inWorking"+str(cnt)+item)

    # switch
    def switch_func(lowerLimit: int, UpperLimit: int) -> str:
        cnt = 0
        file_list = os.listdir(input_dir)
        for item in file_list:
            if "inWorking" in item:
                cnt += 1
        if cnt < 10:
            return "less"
        else:
            return "more"
        
    # convert to mp4
    def convert_to_mp4_func(item: str, output_path: str) -> bool:
        if item[-3:] == "mp4":
            shutil.copy(input_dir+item, output_path+item)
            return True
        target = item.split('.')[0] + ".mp4"
        command = [
            "ffmpeg",
            "-i", input_dir+item,
            "-c:v", "libx264",
            "-c:a", "mp3",
            "-y",
            "-strict", "experimental",
            output_path+target,
        ]
        try:
            file = open("/Users/lll/airflow/debug1.log", "a")
            file.write("convert cmd run\n")
            file.close()
            subprocess.check_output(command)
            file = open("/Users/lll/airflow/debug1.log", "a")
            file.write("convert cmd run\n")
            file.close()
            return True
        except subprocess.CalledProcessError as e:
            print("convert failed:", e)
            return False

    # change video to frames
    def video_to_frame(item: str, input_path:str, output_path: str) -> bool:
        file = open("/Users/lll/airflow/debug1.log", "a")
        file.write("framing\n")
        file.close()
        prefix = item.split('.')[0]
        command = [
            "ffmpeg",
            "-i", input_path+item,
            "-f", "image2",
            "-y",
            output_path+prefix+"%03d.png",
        ]
        try:
            file = open("/Users/lll/airflow/debug1.log", "a")
            file.write("frame cmd run\n")
            file.close()
            subprocess.check_output(command)
            file = open("/Users/lll/airflow/debug1.log", "a")
            file.write("frame cmd end\n")
            file.close()
            return True
        except subprocess.CalledProcessError as e:
            print("cut into frame failed:", e)
            return False

    # watermark
    def add_watermark(item: str, input_path:str, output_path: str) -> bool:
        file = open("/Users/lll/airflow/debug1.log", "a")
        file.write("watermarking\n")
        file.close()
        file_list = os.listdir(input_path)
        prefix = item.split('.')[0]
        for cur_it in file_list:
            if prefix in cur_it:
                command = [
                    "ffmpeg",
                    "-i", input_path+cur_it,
                    "-y",
                    "-vf", "drawtext=fontfile=./tmp/msyhbd.ttc:fontsize=200:x=(w - 10 - text_w):y=10:fontcolor=white:text='lyb':shadowx=4:shadowy=4:alpha=0.5",
                    output_dir+cur_it,
                ]
            try:
                subprocess.check_output(command)
            except subprocess.CalledProcessError as e:
                print("cut into frame failed:", e)
                return False
        return True

    # worker
    def manual_video_func(index: int, cnt: int):
        file_list = os.listdir(input_dir)
        for item in file_list:
            if "inWorking" in item:
                temp = ""
                i = 9
                while i < len(item) and item[i] >= '0' and item[i] <= '9':
                    temp += item[i]
                    i += 1
                num = int(temp)
                if num % cnt == index:
                    temp_dir_mp4 = input_dir+"mp4/"
                    os.makedirs(temp_dir_mp4, exist_ok=True)
                    if convert_to_mp4_func(item, temp_dir_mp4) != True:
                        raise AirflowException("convert to mp4 failed!")
                    target = item.split('.')[0]+".mp4"
                    temp_dir_pic = input_dir+"pic/"
                    os.makedirs(temp_dir_pic, exist_ok=True)
                    if video_to_frame(target, temp_dir_mp4, temp_dir_pic) != True:
                        raise AirflowException("convert to frame failed!")
                    if add_watermark(item, temp_dir_pic, output_dir) != True:
                        raise AirflowException("add watermark failed!")
                    os.makedirs(input_dir+"test")
        
    # finish
    def finish_func():
        file = open("/Users/lll/airflow/debug1.log", "a")
        file.write("end\n")
        file.close()
        file_list = os.listdir(input_dir)
        for item in file_list:
            if os.path.isfile and "inWorking" in item:
                os.remove(input_dir+item)
            else:
                shutil.rmtree(input_dir+item)

    
    count_task = PythonOperator(
        task_id="count",
        python_callable=count_func,
    )

    switch_task = BranchPythonOperator(
        task_id="switch",
        python_callable=switch_func,
        op_args=[0, 10],
    )

    less_task = EmptyOperator(
        task_id="less",
    )

    more_task = EmptyOperator(
        task_id="more",
    )

    worker_task_0_0 = PythonOperator(
        task_id="worker_0_0",
        python_callable=manual_video_func,
        op_args=[0, 2],
    )

    worker_task_0_1 = PythonOperator(
        task_id="worker_0_1",
        python_callable=manual_video_func,
        op_args=[1, 2],
    )

    worker_task_1_0 = PythonOperator(
        task_id="worker_1_0",
        python_callable=manual_video_func,
        op_args=[0, 5],
    )

    worker_task_1_1 = PythonOperator(
        task_id="worker_1_1",
        python_callable=manual_video_func,
        op_args=[1, 5],
    )

    worker_task_1_2 = PythonOperator(
        task_id="worker_1_2",
        python_callable=manual_video_func,
        op_args=[2, 5],
    )

    worker_task_1_3 = PythonOperator(
        task_id="worker_1_3",
        python_callable=manual_video_func,
        op_args=[3, 5],
    )

    worker_task_1_4 = PythonOperator(
        task_id="worker_1_4",
        python_callable=manual_video_func,
        op_args=[4, 5],
    )

    more_complete_task = EmptyOperator(
        task_id="more_completed",
    )

    less_complete_task = EmptyOperator(
        task_id="less_completed",
    )

    finish_task = PythonOperator(
        task_id="finish",
        python_callable=finish_func,
        trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
    )

    count_task >> switch_task >> [less_task, more_task]
    less_task >> [worker_task_0_0, worker_task_0_1] >> less_complete_task
    more_task >> [worker_task_1_0, worker_task_1_1, worker_task_1_2, worker_task_1_3, worker_task_1_4] >> more_complete_task
    [less_complete_task, more_complete_task] >> finish_task

解决方案

1. 修复subprocess的输出与输入处理

ffmpeg处理视频时会产生大量输出，且可能等待stdin输入，导致Airflow worker进程阻塞。将所有subprocess.check_output(command)替换为以下代码：

subprocess.run(
    command,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    stdin=subprocess.DEVNULL,  # 关闭stdin，避免ffmpeg等待输入
    bufsize=0,  # 禁用缓冲，防止输出满缓冲区阻塞
    check=True
)

2. 指定ffmpeg完整路径

Airflow worker的环境变量可能与终端不一致，先在终端执行which ffmpeg获取完整路径（比如/opt/homebrew/bin/ffmpeg），然后替换命令中的ffmpeg为该完整路径，避免因路径问题导致的兼容或权限问题。

3. 禁用ffmpeg交互模式

在ffmpeg命令中添加-nostdin参数，强制关闭stdin交互，彻底避免进程等待用户输入：

command = [
    "/opt/homebrew/bin/ffmpeg",
    "-nostdin",
    "-i", input_dir+item,
    # 其他参数...
]

4. 调整Airflow资源配置

修改airflow.cfg中的以下参数，确保worker有足够资源运行ffmpeg：

worker_memory_limit = 4096（根据机器配置调整，单位MB）
适当调小parallelism和max_active_tasks_per_dag，避免多任务同时运行导致资源耗尽

5. 检查文件权限

确保Airflow worker运行用户对input_dir、output_dir及临时目录有读写权限：

chmod -R 755 /Users/lll/airflow/video_input/
chmod -R 755 /Users/lll/airflow/video_output/

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

Airflow手动触发任务时ffmpeg进程休眠，任务无法完成求助

相关代码 #

1. 修复subprocess的输出与输入处理 #

2. 指定ffmpeg完整路径 #

3. 禁用ffmpeg交互模式 #

4. 调整Airflow资源配置 #