Pytables 高效读取和处理数千个组

Question

这应该很容易，但是我要花很长时间来计算它，而且我想不出任何其他方法来处理我的文件。我会很感激任何想法。

TL;DR:我正在寻找一种快捷方式 for key in store.keys() 并对其中包含的所有数据运行相同的分析 HDF 文件中的每个节点都有 261k 个节点（又名键、组...），这样每个节点的处理独立于其他节点。

我在磁盘上有一个 H5 文件，大小为几百 GB。该文件包含数十万个节点（准确地说是 261k），我想使用相同的方法单独处理它们。每个节点（或组）包含一个表，带有一个日期时间索引和三个浮点数据列。我想为每个表中的每一列计算分位数。基本上，这是 H5 文件的样子（$ ptdump -av 的部分输出）：

/ (RootGroup) ''
    /._v_attrs (AttributeSet), 4 attributes:
     [CLASS := 'GROUP',
      PYTABLES_FORMAT_VERSION := '2.1',
      TITLE := '',
      VERSION := '1.0']
/101P09999 (Group) '101P09999'
   /101P09999._v_attrs (AttributeSet), 15 attributes:
    [CLASS := 'GROUP',
     TITLE := '101P09999',
     VERSION := '1.0',
     data_columns := [],
     encoding := 'UTF-8',
     index_cols := [(0, 'index')],
     info := {1: {'type': 'Index', 'names': [None]}, 'index': {'index_name': 'DATETIME'}},
     levels := 1,
     metadata := {'STATE': 'Georgia', 'LENGTH': 4.86258, 'COUNTRY': 'USA', 'ROAD_NUMBER': 'US-27/GA-1', 'LATITUDE': 34.88279, 'COUNTY': 'Walker', 'LONGITUDE': -85.27023, 'ROAD_NAME': 'Lafayette Rd/Martha Berry Hwy', 'DIRECTION': 'Northbound'},
     nan_rep := 'nan',
     non_index_axes := [(1, ['TTAV', 'TTPC', 'TTFT'])],
     pandas_type := 'frame_table',
     pandas_version := '0.15.2',
     table_type := 'appendable_frame',
     values_cols := ['values_block_0']]
/101P09999/table (Table(2345,), shuffle, blosc(5)) ''
  description := {
  "index": Int64Col(shape=(), dflt=0, pos=0),
  "values_block_0": Float32Col(shape=(3,), dflt=0.0, pos=1)}
  byteorder := 'little'
  chunkshape := (3276,)
  autoindex := True
  colindexes := {
    "index": Index(6, medium, shuffle, zlib(1)).is_csi=False}
  /101P09999/table._v_attrs (AttributeSet), 12 attributes:
   [CLASS := 'TABLE',
    FIELD_0_FILL := 0,
    FIELD_0_NAME := 'index',
    FIELD_1_FILL := 0.0,
    FIELD_1_NAME := 'values_block_0',
    NROWS := 2345,
    TITLE := '',
    VERSION := '2.7',
    index_kind := 'datetime64',
    values_block_0_dtype := 'float32',
    values_block_0_kind := ['TTAV', 'TTPC', 'TTFT'],
    values_block_0_meta := None]

注意：不要注意上述输出中的NROWS 值。此输出仅适用于 2014 年 2 月，我的数据来自 主表中的所有 12 个月。

对，我要做的是将列['TTAV', 'TTPC', 'TTFT'] 中的数据除以组属性metadata['LENGTH']（在本例中为4.86258）。接下来我想根据数据的时间戳把它分成6组。在这 6 个组中的每一个中，我想计算分位数。

我现在的方法是新手：

with pd.HDFStore(store_path, 'r') as store:
    for key in store.keys()
        sens_data = store[key]
        # split_data = split the data into the required groups based on time stamp...
        for data in split_data:
            data /= store.get_storer(key).attrs.metadata['LENGTH']
            perc = split_data.quantile(q=np.arange(0.05, 1, 0.05)).transpose()
            # Create a column to contain sensor name:
            perc[0] = key[1:]
            perc.set_index(0, append=True, inplace=True)
            perc.index.rename(['DATA COL', 'SENS NAME'], inplace=True)
            # Merge perc into a dictionary of dataframes with keys the groups
            # the data was split into, and value a dataframe of appended percs

因此，数据帧的字典最终将如下所示：

In [1]: percentiles['night']
Out[1]:                      0.05   0.10  ...  0.90  0.95
        DATA COL  SENS NAME
                  101P09999  115    118   ...  133   135
        TTAV      101P10000  95     100   ...  120   125
                  101P10001  108    109   ...  111   113
                     ...
                  101P09999  110    112   ...  133   135
        TTPC      101P10000  115    118   ...  133   135
                  101P10001  115    118   ...  133   135
                     ...
                  101P09999  115    118   ...  133   135
        TTFT      101P10000  115    118   ...  133   135
                  101P10001  115    118   ...  133   135
                     ...                  ...

（请原谅我没有为其余行输入随机数据。此外，虽然我在示例 DataFrame 中输入了 int 值，但这些值实际上是 float32，如 ptdump 输出所示.) 但这就是上面代码末尾所有组（晚上、早上、下午......）的样子。所以基本上，每个['TTAV', 'TTPC', 'TTFT']，都会有261k行，每行对应一个传感器。

现在很明显，for key in store.keys() 不是要走的路。 （仅检索所有键大约需要 15 分钟！）我发现了无数示例，其中仅检索和处理一组和表的数据，但没有帮助处理所有组。有什么想法吗？并行读取是可以的，虽然我还没有让它工作（当我使用密钥访问不同进程中节点内的数据时，它会引发 UnImplimented 错误。也许我还需要将存储传递给函数）。然而，最大的问题是for key in store.keys() 需要 15 分钟才能返回所有键的列表。 （请注意，我在处理时不需要所有键的列表，我很乐意在解析文件时获取数据。）

Answer 1

生活依旧美好。我想出了如何同时从 HDF 文件中读取。所以现在等待 15 分钟来生成所有键的列表似乎并不算太​​糟糕。基本上，这就是现在正在发生的事情：

def reader(key)
    with pd.HDFStore(store_path, 'r') as store:
        sens_data = store[key]
        # And other awesome stuff here to manipulate the data

    # Closing with returning the analyzed data, in my case, percentile computation:
    return dict_of_dfs

if __name__ == '__main__':
    with Pool(processes=4) as pool:
        res_pool = pool.map(reader, sens_names) # I populate sens_names by store.keys() up above.

for key in dict_of_dfs:
    dict_of_dfs[key] = dict_of_dfs[key].append(d[key] for d in res_pool)

瞧！比单线程快得多！

---

附言。我仍然想知道一些关于完全避免store.keys() 的事情。