Python如何实现异步的数据清洗 pipeline_基于协程的任务流设计

Python如何实现异步的数据清洗 pipeline_基于协程的任务流设计

asyncio.run() 不能直接套用在已有事件循环中

很多开发者初次尝试异步清洗时，会习惯性地在代码里写上 asyncio.run(clean_pipeline())。结果呢？一旦代码跑在Jupyter、FastAPI或者Django这类已经启动了事件循环的环境里，立马就会抛出 RuntimeError: asyncio.run() cannot be called from a running event loop。这其实不是逻辑错误，而是调用时机没搞对。

具体该怎么操作？记住下面几点：

• 如果你的代码是独立的脚本或进程入口，放心使用 asyncio.run()；
• 但如果身处一个已经运行着事件循环的上下文（比如FastAPI的一个路由处理函数里），直接 await clean_pipeline() 才是正解；
• 如果环境不确定，一个实用的技巧是先用 asyncio.get_event_loop().is_running() 判断一下，再决定用 run() 还是 create_task()。

map() 和 async for 不兼容，别硬套同步惯性思维

习惯了 pandas.DataFrame.map() 或者Python内置的 map() 函数，很容易下意识写出 map(async_clean, rows) 这样的代码。但问题来了：这行代码返回的是一堆协程对象（coroutine objects），它们本身并不是可等待的（awaitable）。如果你试图去 await 这个结果，就会遇到 TypeError: object XXX can‘t be used in ’await‘ expression。

正确的打开方式是这样的：

• 先用列表推导式构造协程列表：[async_clean(row) for row in rows]；
• 然后统一交给 await asyncio.gather(*coro_list) 来并发执行；
• 如果数据量特别大，担心内存吃不消，可以引入 asyncio.Semaphore(10) 来控制并发度，配合 async for 和 async with semaphore: 来分批处理；
• 核心原则就一条：在异步世界里，忘掉同步的 map() 函数，它既不支持 await，也不会返回你想要的结果。

asyncpg / aiomysql 的连接池必须显式关闭，否则 pipeline 会卡死

数据清洗流程里，经常需要查询维度表或者把结果写回数据库。用了 asyncpg.create_pool() 创建连接池之后，如果只调用 pool.fetch() 而不妥善管理池子的生命周期，麻烦就来了。第二次运行pipeline时，程序可能会莫名其妙地卡在获取连接这一步。日志里风平浪静，没有错误，CPU占用也很低，但就是没有响应——这通常是连接池耗尽或者资源未被释放的典型症状。

避免这个坑，得遵循几个最佳实践：

• 把连接池当作异步上下文管理器来用：async with create_pool(...) as pool:，让Python自动处理资源的获取和释放；
• 如果需要在多个清洗任务间复用同一个池子，记得用 asyncio.Lock() 来包裹 pool.acquire()，防止并发场景下的资源争抢；
• 千万别在一个协程里调用 pool.close() 之后，还试图继续使用这个池子，这会直接触发 InvalidStateError；
• 本地调试时，可以加一句 print(f"Pool size: {pool._size}, free: {pool._free}") 来直观地观察连接池的状态。

Pydantic v2 的 model_validate() 不支持 await，但 model_validate_json() 可以间接配合

清洗完的数据，用Pydantic做结构化校验是常见操作。但如果你直接写 await User.model_validate(row)，立刻就会收到 TypeError: object ModelMetaclass can‘t be used in ’await‘ expression。原因很简单：model_validate() 本身是一个同步方法。有些人可能会想，那我把它包在一个 async def 函数里总行了吧？这其实只是换了个包装，并没有解决潜在的IO阻塞问题。

正确的处理思路需要分情况讨论：

• 如果校验过程本身不涉及任何IO操作（大多数情况），那就保持同步调用：User.model_validate(row)；
• 如果校验前需要从远程加载schema或规则（比如从一个JSON Schema URL获取配置），那么应该用 httpx.AsyncClient().get() 这类异步客户端去获取配置，获取完成后再进行同步校验；
• 如果数据行（row）本身来自异步读取（比如用 aiofiles.open() 读的文件），务必确保先通过 await 拿到完整的字符串，再传给 model_validate_json()；
• 归根结底，记住一个原则：不要给Pydantic的校验方法加 await，它们不是协程。

说到底，构建一个高效的协程pipeline，核心要义不在于“把所有函数都变成async”，而在于精准地识别IO密集点并实现真正的挂起，在CPU密集处适时让出控制权，同时确保资源（尤其是连接池）的生命周期清晰可控。经验表明，最容易出问题的地方，往往不是语法细节，而是连接池与事件循环状态之间那些微妙的耦合点。