Pytables / Pandas：组合（读取？）按行分割的多个HDF5存储

在“一次编写，多次读取”工作流程中，我经常使用FastExport实用程序解析从Teradata转储的大型文本文件（20GB-60GB），并使用Pandas将它们加载到Pytables中 . 我正在使用多处理来分块文本文件并将它们分发到不同的进程，以便根据每行大约5MM的行数来编写.H5文件，以支持并行写入 . 对于并行编写多个hdf5文件大约12分钟这相当快，相比之下，为25MM行x64列编写单个hdf5文件需要22分钟 .

%timeit -n 1 write_single_hdf_multiprocess()
1 loops, best of 3: 22min 42s per loop

%timeit -n 1 write_multiple_hdf_multiprocess()
1 loops, best of 3: 12min 12s per loop

对于按行分割多个h5文件的情况，我最终会有多个文件具有相同的结构，我希望在单个h5file根/ data / table中组合

要测试组合功能，以下是代码段：

import tables as tb
import pandas as pd

tb.setBloscMaxThreads(15)
store =pd.HDFStore('temp15.h5',complib='blosc')

filenames=['part_1.h5','part_2.h5','part_3.h5','part_4.h5','part_5.h5']

for f in filenames:
    s=pd.HDFStore(f)
    df=s.select('data')
    store.append(key='data',value=df,format='t',chunksize=200000)

store.close()

这是％timeit结果：

1 loops, best of 3: 8min 22s per loop

这基本上占用了我通过并行编写多个h5文件获得的大部分时间 . 我有两个问题：

• 有没有办法更有效地组合（追加）具有相同表格格式的h5文件？（SQL联盟就像功能一样） . 我试过this SO但是无法让它附加表格 .

• 如果没有，在大多数查询从所有列的位置中选择时，在行上拆分是否合理？我正在考虑编写一个map / combine函数，它将查看表的所有部分，以便从查询中进行选择 . Pandas select_as_multiple()函数执行此操作以基于列进行拆分 .

Update Based on Jeff's Suggestions:

在合并前文件写入过程中删除索引和压缩的很棒的调用 . 删除索引，压缩并将每个预合并文件的最大行数设置为1MM行：

%timeit -n 1 write_multiple_hdf_multiprocess()
1 loops, best of 3: 9min 37s per loop

这比以前快2分多一点，速度和解析数据的速度差不多 . 将数据列设置为所需的字段后（在我的情况下为3）：

for f in filenames:
    s=pd.HDFStore(f)
    df=s.select('data')
    dc=df.columns[1:4]
    store.append(key='data',value=df,format='t',data_columns=dc)

这比以前慢了大约2分钟： 1 loops, best of 3: 10min 23s per loop . 从上面的代码中删除压缩后，我得到 1 loops, best of 3: 8min 48s per loop （几乎与第一次尝试压缩而没有数据列索引相同） . 为了让您了解压缩的效果，未压缩存储大约为13.5GB，而使用 blosc 的压缩版本大约为3.7GB .

总之，我的进程需要 18 minutes 15 seconds 来创建一个合并的未压缩的hdf5文件 . 与单个文件写入（压缩）相比，这个速度大约为 4 minutes 7 seconds .

这让我想到了我的问题的第二部分，如果我不合并文件并使用合并前文件以 Map /组合方式处理，那么这可能是一种合理的方法吗？我应该如何考虑实施这个？

对于完整的披露，我在Pandas版本 0.12.0 ，Pytables版本 3.0.0 和我的数据处理工作流程如下（伪代码）：

def generate_chunks_from_text_file(reader,chunksize=50000):
    """ generator that yields processed text chunks """

    for i, line in enumerate(reader.readlines()):
        ----process data and yield chunk -----


def data_reader(reader,queue):
    """ read data from file and put it into a queue for multiprocessing """

    for chunk in self.generate_chunks_from_text_file(reader):
        queue.put(chunk) # put data in the queue for the writer

def data_processor(queue,filename,dtype,min_size):
    """" subprocess that reads the next value in the queue and writes hdf store. """

    store=pd.HDFStore(filename)

    while True:

        results = queue.get()
        array=np.array(results,dtype=dt) # convert to numpy array
        df = pd.DataFrame(array) #covert to pandas array

        store.append(key='data', value=df, format='t', min_itemsize=dict(min_size), data_columns=[],index=False)
    store.close()
        ----when queue exhausts - break-----