Python Pandas 线程读取 read_fwf答案

【问题标题】：Python Pandas Threading reads read_fwfPython Pandas 线程读取 read_fwf
【发布时间】：2019-05-12 01:20:45
【问题描述】：

这里是新手 python/pandas 用户。我一直在尝试在 read_fwf 中使用 chunksize arg 并迭代变量的 value_counts。我写了一个函数来传递参数，例如文件迭代器和变量来解析和计数。我希望并行化这个函数，并能够同时将 2 个文件读入同一个函数中。它似乎确实有效......但是，我遇到了意想不到的减速。线程同时完成，但一个似乎正在减慢另一个（IO 瓶颈？）。通过按顺序运行而不是并行运行函数（324 秒对 172 秒），我得到了更快的时间。想法？我执行错了吗？我已经尝试过多进程但启动映射错误，我无法腌制文件迭代器（read_fwf 的输出）。

testdf1=pd.read_fwf(filepath_or_buffer='200k.dat',header=None,colspecs=wlist,names=nlist,dtype=object,na_values=[''],chunksize=1000)    
testdf2=pd.read_fwf(filepath_or_buffer='200k2.dat',header=None,colspecs=wlist,names=nlist,dtype=object,na_values=[''],chunksize=1000)

def tfuncth(df,varn,q,*args):
    td={}
    for key in varn.keys():
        td[key]=pd.Series()
    for rdf in df:
        if args is not None:
            for arg in args:
                rdf=eval(f"rdf.query(\"{arg}\")")
        for key in varn.keys():
            ecode=f'rdf.{varn[key]}.value_counts()'
            td[key]=pd.concat([td[key],eval(ecode)])
            td[key]=td[key].groupby(td[key].index).sum()
    for key in varn.keys():
        td[key]=pd.DataFrame(td[key].reset_index()).rename(columns={'index':'Value',0:'Counts'}).assign(Var=key,PCT=lambda x:round(x.Counts/x.Counts.sum()*100,2))[['Var','Value','Counts','PCT']]
    q.put(td)

  bands={
        '1':'A',
        '2':'B',
        '3':'C',
        '4':'D',
        '5':'E',
        '6':'F',
        '7':'G',
        '8':'H',
        '9':'I'
    }
    vdict={
        'var1':'e1270.str.slice(0,2)',
        'var2':'e1270.str.slice(2,3)',
        'band':'e7641.str.slice(0,1).replace(bands)'
    }

    my_q1=queue.Queue()
    my_q2=queue.Queue()
    thread1=threading.Thread(target=tfuncth,args=(testdf1,vdict,my_q1,flter1))
    thread2=threading.Thread(target=tfuncth,args=(testdf2,vdict,my_q2))
    thread1.start()
    thread2.start()

更新：

经过大量阅读这也是我得出的结论。这是非常简化的结论，我敢肯定，如果有人知道，请通知我。

Pandas 不是一个完全多线程友好的包
- 显然有一个名为“dask”的包，它复制了很多 pandas 函数。所以我会调查一下。
在许多情况下，Python 并不是真正的多线程兼容语言案例
- Python 受其编译器约束。在纯 python 中，它由 GIL 解释和绑定，一次只能执行一个线程
  - 可以分离出多个线程，但只能并行处理非 CPU 绑定的函数。
  - 我的代码用 IO 和 CPU 包装。简单的 IO 可能是并行运行的，但在等待处理器执行时被耽搁了。
  - 我计划通过编写仅 IO 操作并尝试线程化来对此进行测试。
- 可以使用在线程上没有全局解释器锁 (GIL) 的不同编译器编译 Python。
  - 因此，诸如“dask”之类的包可以利用多线程。

【问题讨论】：

标签： python pandas

【解决方案1】：

我确实设法通过使用 multiprocessing 包让它工作并解决了我的问题。我遇到了两个问题。
1) multiprocessing 包不兼容 Juypter Notebook

和

2) 你不能腌制熊猫阅读器的句柄（传递给进程的多处理腌制对象）。

我通过在 Notebook 环境之外进行编码来修复 1，并通过将打开分块文件所需的参数传递给每个进程并让每个进程开始自己的分块读取来修复 2。

完成这两件事后，我的速度比连续运行提高了 60%。

【讨论】：