Pandas HDFStore只读连接下无需重开获取新增追加数据的方法

结论 #

可以无需重新打开HDFStore实现新增行读取，核心是主动刷新底层PyTables的节点元数据和缓存，绕过pandas默认的内存缓存机制。

原理说明 #

pandas的HDFStore基于PyTables实现，PyTables默认会将已打开文件的节点结构、已读取数据块缓存在内存中，外部进程追加写入的内容不会主动触发已打开只读连接的元数据同步，因此默认无法读取到新增行。

实现方案 #

核心操作 #

每次轮询查询前，强制刷新目标存储节点的缓存和磁盘元数据即可，不需要关闭重建整个HDFStore连接。

完整代码示例 #

import pandas as pd
import time
import signal

# 全局存储HDFStore实例，用于退出时安全关闭
hdf_store = None
# 你存储数据对应的HDF键名
DATA_KEY = "/your_data_key"

def safe_exit(signum, frame):
    """退出信号处理，保证存储正常关闭"""
    if hdf_store is not None and hdf_store.is_open:
        hdf_store.close()
    exit(0)

if __name__ == "__main__":
    # 注册中断、终止信号的处理逻辑
    signal.signal(signal.SIGINT, safe_exit)
    signal.signal(signal.SIGTERM, safe_exit)
    
    # 只读模式打开HDF文件，支持多进程并行读取
    hdf_store = pd.HDFStore("your_data.h5", mode="r")
    # 记录上一次读取的最后一行位置
    last_offset = 0

    while True:
        # 核心步骤1：刷新目标节点缓存，同步最新元数据
        hdf_store.get_node(DATA_KEY)._f_flush()
        # 核心步骤2：刷新整个文件的底层句柄缓存，适配多节点写入场景
        hdf_store._handle.flush()

        # 获取当前节点总行数
        current_rows = hdf_store.get_storer(DATA_KEY).nrows
        if current_rows > last_offset:
            # 仅读取新增部分，避免全表扫描浪费IO资源
            new_data = hdf_store.select(DATA_KEY, start=last_offset, stop=current_rows)
            # 此处填写你的业务处理逻辑
            print(f"获取到{len(new_data)}条新增记录")
            # 更新读取偏移量
            last_offset = current_rows

        # 按需调整轮询间隔，避免空转占用CPU
        time.sleep(1)

注意事项 #

• 该方案仅适配单进程写入、多进程只读读取的场景，不要在多进程同时写入的场景中使用，避免文件损坏。
• 不要调用hdf_store.flush()：该方法仅针对写入模式的连接生效，只读模式下调用无意义且可能触发警告。
• 针对你150个HDF文件的场景，可对每个打开的HDFStore实例单独执行上述刷新逻辑即可，资源开销远低于每次重新打开文件。

备选兼容方案 #

如果使用的旧版本PyTables上述刷新逻辑不生效，可使用更低开销的节点重开方法替代，无需关闭整个存储：

# 仅重开目标节点，加载最新元数据，开销远低于重建整个HDFStore连接
hdf_store.get_node(DATA_KEY)._f_reopen()

内容的提问来源于stack exchange，提问作者Aram Darakchian