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

Question

在“一次写入，多次读取”工作流程中，我经常使用 FastExport 实用程序解析从 Teradata 转储的大型文本文件 (20GB-60GB)，并使用 Pandas 将它们加载到 Pytables 中。我正在使用多处理来分块文本文件并将它们分发到不同的进程，以编写基于每个 5MM 左右的行数拆分的 .H5 文件，以支持并行写入。并行写入多个 hdf5 文件大约需要 12 分钟，相比之下，为 25MM 行 x 64 列写入单个 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 root/data/table

为了测试组合功能，这里是代码sn-p：

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 文件获得的大部分时间。我有一个两部分的问题：

1. 有没有办法更有效地组合（附加）具有相同表格格式的 h5 文件？（类似 SQL 联合的功能）。我尝试了this SO，但无法将其附加到表格中。

2. 如果不是，当大多数查询都是从所有列的位置选择时，拆分行是否合理？我正在考虑编写一个 map/combine 函数，该函数将查看表的所有部分以从 where 查询中进行选择。 Pandas select_as_multiple() 函数可以根据列进行拆分。

---

根据 Jeff 的建议进行更新：

删除合并前文件写入过程中的索引和压缩非常重要。删除索引、压缩并将每个预合并文件的最大行数设置为 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/combine 方式处理怎么办，这是否是一种合理的处理方式这个？我应该如何考虑实现这一点？

为了全面披露，我使用的是 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-----

Answer 1

我做了一个非常相似的，split-process-combine 方法，使用多个进程创建中间文件，然后使用单个进程合并生成的文件。以下是获得更好性能的一些提示：

• 在编写文件时通过传递 index=False 关闭索引，有关文档，请参阅 here。我相信PyTables 会增量更新索引，在这种情况下这是完全没有必要的（因为您将在之后合并它们）。仅索引最终文件。这应该会加快写作速度。

• 您可能会考虑更改默认索引方案/级别，具体取决于您的查询是什么（假设您遵循以下几点建议不要创建太多数据列）。

• 与此类似，在写入预合并文件时不要创建压缩文件，而是在写入索引文件后（处于未压缩状态）创建它，因此这是您的最后一步.请参阅文档here。此外，在使用 ptrepack 时传递 --chunkshape=auto 非常重要，它会重新计算 PyTables 的块大小（例如，在单个块中读取/写入多少数据），因为它将考虑整个表。

• RE 压缩，YMMV 在此处可能会有所不同，具体取决于您的数据实际压缩的程度以及您正在执行的查询类型。我有一些类型的数据，我发现根本不压缩会更快，即使理论上它应该更好。你只需要进行实验（尽管我总是使用blosc）。 Blosc 只有 一个 压缩级别（1-9 级打开或 0 级关闭）。所以改变它不会改变任何东西。

• 我按索引顺序合并文件，基本上是通过将预合并文件的一个子集读入内存（一个常数，只使用一个常数的内存），然后一个接一个地追加它们到最终文件。 （不是 100% 确定这会有所作为，但似乎效果很好）。

• 您会发现大部分时间都花在creating 索引上。

• 此外，只索引您实际需要的列！确保在写入每个文件时指定data_columns=a_small_subset_of_columns。

• 我发现写很多小文件比较好，然后合并创建一个大文件，而不是写几个大文件，但是这里是YMMV。 （例如，说 100 个 100MB 预合并文件以产生 10GB 文件，而不是 5 个 2GB 文件）。虽然这可能是我的处理管道的一个功能，因为我倾向于在处理而不是实际写入方面遇到瓶颈。

• 我没用过，但听说过关于使用 SSD（售出状态驱动器）的惊人事情，即使它对于这类事情来说相对较小。使用 1 可以获得一个数量级的加速（压缩可能会改变这个结果）。