Pandas+multiprocessing处理大文件时内存溢出问题

from pandas import read_csv
from multiprocessing import Pool

def _process(_chunk):
    # 分块处理逻辑...
    pass

if __name__ == '__main__':
    pool_size = 4
    pool = Pool(processes=pool_size)
    reader = read_csv(SOURCE_FILE, sep='t', header=None, usecols=[1], skiprows=1, iterator=True, chunksize=10000)

    for chunk in reader:
        pool.apply_async(_process, [chunk])

    pool.close()
    pool.join()

如上是一段简单的单机多进程处理较大CSV文件的Python代码，笔者在处理超过30GB的CSV文件时发现，主进程占用内存一直在增长，子进程内存占用相对稳定，最终把20G内存的云虚拟机跑挂了，只能重启…
无奈查了大量资料，也看了很多Python内存溢出、内存泄露相关的调试方法（pyrasite官网有惊喜， hahaha），什么gdb psutil objgraph，最后无意间发现了问题所在。
话不多说，先上解决方案：

from pandas import read_csv
from multiprocessing import Pool

def _process(_chunk):
    # 分块处理逻辑...
    pass

if __name__ == '__main__':
    pool_size = 4
    pool = Pool(processes=pool_size)
    reader = read_csv(SOURCE_FILE, sep='t', header=None, usecols=[1], skiprows=1, iterator=True, chunksize=10000)

    for chunk in reader:
    #####################################################
    # 根据环境自己调整阈值，此处设置的pool_size * 200
        result = pool.apply_async(_process, [chunk])
        if len(pool._cache) > pool_size * 200:
            result.wait()
    #####################################################

    pool.close()
    pool.join()

apply_async是进程池异步处理方法，往任务队列中新增任务后即退出，非阻塞；所以这里会带着任务参数chunk一直往任务队列里追加，追加的速度是远快于块数据处理的速度，所以队列中积压的任务越来越多，直至无内存可分配，boom；
Pool中_cache变量保存着待执行任务的ApplyResult，所以len(pool._cache)可以得到待执行任务数量，此时超过一定数量，我们让当前提交任务进入等待状态，等待任务全部执行完成再进行任务提交，这样内存不会持续增长，而是会增长减少增长减少，不会超过峰值；不过这个方案还有个缺点，就是wait时需要等待积压的所有任务完成才会开始，更合理的是完成一个新增一个这样，节约时间提高效率， TODO…