python - pandas 中的并行 read_table

Question

有没有办法并行调用 read_table()？就我而言，由于日期解析，它受到 CPU 限制。我看不出有什么方法可以通过阅读文档来实现这一目标。唯一想到的是分割输入文件，并行调用 read_table，然后连接数据帧。

Answer 1

这将并行读取 CSV 文件并将它们连接起来。烦人的一点是它无法处理 numpy 类型，因此它无法解析日期。我一直在努力解决同样的问题，但到目前为止，像 execnet 这样的库似乎无法处理非内置类型。这就是为什么我在发送之前将 DataFrames 转换为 json 的原因。它将类型剥离为基本的 Python 类型。

编辑:如果您需要解析日期，也许更明智的方法是远程读取 CSV 文件，解析日期并将其另存为 pickle 到硬盘。然后你可以在主进程中读取pickle文件并将它们连接起来。我还没有尝试过这样做是否会提高性能。

remote_read_csv.py

import cPickle as pickle

if __name__ == '__channelexec__':
    reader = pickle.loads(channel.receive())

    for filename in channel:
        channel.send(reader(filename).to_json())

下面使用了上面的模块。我在 IPython 中测试了它。

from pandas import DataFrame, concat, read_csv, read_json
from numpy import random
import execnet
import remote_read_csv
import cPickle as pickle
import itertools
import psutil

### Create dummy data and save to CSV

def rdf():
    return DataFrame((random.rand(4, 3) * 100).astype(int))

d1 = rdf()
d2 = rdf()
d3 = rdf()

dfsl = [d1, d2, d3]
names = 'd1.csv d2.csv d3.csv'.split()
for i in range(3):
    dfsl[i].to_csv(names[i])

### Read CSV files in separate threads then concatenate

reader = pickle.dumps(read_csv)

def set_gateways(remote_module, *channel_sends):
    gateways = []
    channels = []
    for i in range(psutil.NUM_CPUS):
        gateways.append(execnet.makegateway())
        channels.append(gateways[i].remote_exec(remote_module))
        for send in channel_sends:
            channels[i].send(send)
    return (gateways, channels)

def para_read(names):
    gateways, channels = set_gateways(remote_read_csv, reader)
    mch = execnet.MultiChannel(channels)
    queue = mch.make_receive_queue()
    channel_ring = itertools.cycle(mch)
    for f in names:
        channel = channel_ring.next()
        channel.send(f)
    dfs = []
    for i in range(len(names)):
        channel, df = queue.get()
        dfs.append(df)

    [gw.exit() for gw in gateways]
    return concat([read_json(i) for i in dfs], keys=names)

para_read(names)