python - 在 HDF5 数据集中查找唯一列

Question

我正在使用 HDF5 来存储非常大的 uint8s 数据集 (400 x 121000000)。列中存在大量冗余(97% 的列不是唯一的)。我需要有效地合并重复的列。这意味着我需要删除重复的列，同时存储元数据以记住合并了哪些列。

我目前正在使用 Python 和 h5py，但如果有人有高效的 C++ 解决方案，我可以简单地使用 boost::python 来实现它。

我当前的解决方案是将数据集的 block 加载到 NumPy 数组中，并使用 dictionary 来存储唯一的列和元数据。

注意:可以找到HashableNDArray 类here .我刚刚重命名了它。

def find_column_redundancy(dataset):
    n_columns = dataset.shape[1]
    block_size = 500000
    n_blocks = int(ceil(float(n_columns) / float(block_size)))

    d = {}
    analysed_column_count = 0
    for block in xrange(n_blocks):
        block_offset = block*block_size
        block_data = dataset[:, block_offset : block_offset+block_size]
        for i in xrange(block_data.shape[1]):
            hashable_array = HashableNDArray(np.ascontiguousarray(block_data[:, i]))
            d[hashable_array] = np.append(d.get(hashable_array, np.array([], dtype=np.int32)), block_offset + i)
            analysed_column_count += 1

    return d

一旦我遍历了所有列，我将返回一个字典，我用它来编写一个新的HDF5数据集并删除了冗余。

我需要帮助；这不是最优的!

谢谢!

Answer 1

我用 kernprof 做了一些分析并优化了我的代码。

• 最大的瓶颈是 HashableNDArray 对象的实例化。我发现通过将 numpy 数组设置为只读，我可以在不使用包装器类的情况下散列它们的数据缓冲区。此外，将缓冲区数据提取为字符串似乎可以实现更快的哈希处理。为了恢复列数据，我使用了 np.frombuffer(dict_key, dtype=np.uint8)。

• 我还通过将字典替换为 defaultdict 获得了小幅加速。并消除 try/except block 。

• 由于我的数据只包含二进制值，我发现使用 np.packbits在列上允许在存储键时将内存节省 8 倍，并且仍然允许匹配相同的列。您唯一需要记住的是使用 np.unpackbits是列的实际长度，因为 numpy 用尾随 0 填充不完整的字节。

最后，我微调了 block_size 以使用最大可用内存量。这允许稍微更长的磁盘读取和更好的 CPU 使用率。

此函数过去在我的数据上运行约 18 小时，现在运行约 0.5 小时!

def find_column_redundancy(dataset):
    n_columns = dataset.shape[1]
    block_size = 10000000
    n_blocks = int(ceil(float(n_columns) / float(block_size)))

    d = defaultdict(list)
    analysed_column_count = 0
    for block in xrange(n_blocks):
        block_offset = block*block_size
        block_data = dataset[:, block_offset : block_offset+block_size]
        block_data = np.asfortranarray(block_data)
        block_data = np.packbits(block_data, axis=0)
        block_data.flags.writeable = False
        for i in xrange(block_data.shape[1]):
            d[block_data[:, i].data[:]].append(block_offset + i)
            analysed_column_count += 1

        print float(analysed_column_count)/n_columns*100, "% completed. Dictionnary has", len(d), "items."

return d