Airflow处理MongoDB数据时内存过载任务停滞问题求助

DevOps 小满

2026-10-6

问题根因分析

核心概念错误：DAG顶层代码执行逻辑错误
Airflow调度器会按固定周期（默认30s）全量解析所有DAG文件，你将MongoDB查询、游标遍历、动态任务生成的逻辑直接写在DAG顶层作用域，每次解析都会触发以下操作：创建MongoDB连接、拉取指定数量的文档、为每个文档生成PythonOperator对象、维护任务依赖关系。这些操作生成的大量内存对象不会被主动回收，随时间持续堆积直接导致内存泄漏。
配置参数不合理
1. 调度间隔与任务处理时长不匹配：单轮任务需要2分钟完成，你设置了每分钟调度一次，即使开启了max_active_runs=1，调度器也会持续生成排队中的DAG Run实例，占用元数据库存储和调度器内存。
2. 重试次数过高：retries设置为30次，单次重试间隔5分钟，任务失败后会在未来150分钟内持续生成重试任务实例，进一步加剧资源占用。
3. MongoDB游标未释放：你使用了no_cursor_timeout=True的游标，且没有显式关闭逻辑，每次DAG解析生成的游标都会长期占用MongoDB连接和本地内存，加重资源泄漏。
任务设计不合理
你将完整文档对象直接作为op_kwargs传入Operator，文档内容会被序列化存储到Airflow元数据库，同时在DAG解析阶段就会加载全量文档内容到内存，进一步加速内存占用增长。

优化解决方案

重构DAG逻辑，将数据查询放到任务内部
不要在DAG顶层操作数据库，所有业务逻辑都放到Operator执行阶段，推荐使用Airflow 2.0+的动态任务映射功能实现单文档并行处理，示例代码如下：

from airflow.decorators import dag, task
from airflow.operators.dummy import DummyOperator
from datetime import datetime, timedelta

DEFAULT_ARGS = {
    "owner": "airflow",
    "retries": 3, # 降低重试次数到合理范围
    "retry_delay": timedelta(minutes=5),
    "depends_on_past": False,
}

@dag(
    dag_id="mongo_process_upload",
    start_date=datetime(2021, 3, 5, 10, 30),
    schedule_interval="*/2 * * * *", # 调整调度间隔匹配处理时长
    default_args=DEFAULT_ARGS,
    max_active_runs=1,
    catchup=False,
)
def process_dag():
    start = DummyOperator(task_id='start')
    end = DummyOperator(task_id='end')

    @task
    def fetch_listings():
        # 将MongoDB查询放到任务内部，仅在DAG运行时执行
        db = get_mongo_db().get_database(CF["base"])
        collection_spider = db[CF["spider"]]
        cursor_spider = (
            collection_spider.find(CF["spider_filter"], {"_id": 1}) # 仅拉取需要的字段
                .limit(CF["number"])
                .skip(CF["offset"])
                .sort('updated_at', ASCENDING)
        )
        doc_ids = [doc["_id"] for doc in cursor_spider]
        cursor_spider.close() # 显式关闭游标释放资源
        return doc_ids

    @task
    def publish_pics(doc_id):
        # 单个文档处理逻辑，运行时再查询完整文档内容
        db = get_mongo_db().get_database(CF["base"])
        collection_spider = db[CF["spider"]]
        document = collection_spider.find_one({"_id": doc_id})
        # 原有图片处理、上传逻辑
        ...

    # 动态映射：将拉取到的文档ID列表映射为多个并行的处理任务
    doc_ids = fetch_listings()
    start >> publish_pics.expand(doc_id=doc_ids) >> end

dag = process_dag()

调整调度配置
如果需要更接近实时的处理，不要用固定频率调度，改用MongoDB Sensor监听新数据生成，避免无效的空跑调度。
优化Executor配置
确认使用LocalExecutor而非默认的SequentialExecutor，调整以下核心配置控制并发量，避免超出硬件负载：

parallelism：全局最大并行任务数，建议设置为16（8核2倍）
dag_concurrency：单DAG最大并行任务数，建议设置为8
worker_concurrency：单Worker进程最大并行任务数，建议设置为8

额外优化建议

开启DAG解析缓存，减少重复解析的资源消耗
定期清理Airflow元数据库中老旧的任务实例、DAG Run记录，降低元数据库压力

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