多进程并发写入同一HDF5文件遇h5py锁错误问题咨询

我之前也碰到过一模一样的问题，这本质上是HDF5的文件锁机制在起作用——它默认不允许多个进程同时以写入模式打开同一个文件，所以你看到的OSError: Unable to lock file, errno = 11是完全符合预期的；而处理异常时触发的第二个OSError，大多是因为文件锁处于异常状态，导致后续操作无法正常执行。

下面给你几个可行的解决方案，按推荐程度排序：

1. 使用HDF5 MPI并行驱动（官方推荐的多进程写入方案） #

如果你的运行环境支持MPI（比如集群或者本地安装了mpi4py），这是最可靠的方案。h5py可以配合mpi4py，让多个进程通过MPI协作文书同一个HDF5文件，底层由HDF5的并行I/O驱动处理锁和数据同步。

示例代码：

from mpi4py import MPI
import h5py

# 初始化MPI通信
comm = MPI.COMM_WORLD
rank = comm.Get_rank()
total_processes = comm.Get_size()

# 所有进程必须以相同的模式和驱动打开文件
with h5py.File('shared_data.h5', 'w', driver='mpio', comm=comm) as h5file:
    # 创建一个可以被多进程写入的数据集
    dataset = h5file.create_dataset(
        'multi_process_data',
        shape=(total_processes, 1000),  # 每个进程写一行
        dtype='float32'
    )
    # 每个进程写入自己负责的切片，避免数据冲突
    dataset[rank, :] = [rank * 0.1 for _ in range(1000)]

运行时需要用MPI启动，比如mpiexec -n 4 python your_script.py（4是进程数）。

2. 单进程作为写入服务端（适合无MPI的环境） #

如果你的环境不支持MPI，可以用“主进程负责写入，子进程负责生成数据并传递给主进程”的模式。通过Python的multiprocessing.Queue实现进程间数据传递，完全规避多进程同时打开文件的问题。

示例代码：

import multiprocessing as mp
import h5py

def data_worker(queue, process_id):
    """子进程：生成数据并发送到队列"""
    # 模拟业务逻辑生成数据
    generated_data = [process_id for _ in range(500)]
    queue.put(('main_dataset', process_id, generated_data))

def hdf5_writer(queue, total_processes):
    """主进程：负责打开文件并写入所有子进程的数据"""
    with h5py.File('shared_data.h5', 'w') as h5file:
        # 预先创建好数据集
        dataset = h5file.create_dataset(
            'main_dataset',
            shape=(total_processes, 500),
            dtype='int'
        )
        received_count = 0
        while received_count < total_processes:
            msg = queue.get()
            if not msg:
                continue
            dset_name, idx, data = msg
            dataset[idx, :] = data
            received_count += 1

if __name__ == '__main__':
    process_num = 4
    # 创建进程间通信队列
    data_queue = mp.Queue()

    # 启动写入进程
    writer_proc = mp.Process(target=hdf5_writer, args=(data_queue, process_num))
    writer_proc.start()

    # 启动所有数据生成进程
    worker_procs = []
    for i in range(process_num):
        proc = mp.Process(target=data_worker, args=(data_queue, i))
        worker_procs.append(proc)
        proc.start()

    # 等待所有子进程完成
    for proc in worker_procs:
        proc.join()

    # 通知写入进程结束
    writer_proc.join()

3. 临时文件合并（适合数据可拆分的场景） #

如果你的数据可以拆分成独立的部分，每个进程写入自己的临时HDF5文件，所有进程完成后再合并成一个文件。这个方案不需要进程间通信，实现简单，但需要额外的磁盘空间存储临时文件。

示例步骤：

• 每个进程写入temp_data_{process_id}.h5
• 所有进程完成后，用h5py遍历所有临时文件，将数据合并到最终文件中：

import h5py
import glob

# 合并临时文件
with h5py.File('final_data.h5', 'w') as final_file:
    temp_files = glob.glob('temp_data_*.h5')
    for idx, temp_path in enumerate(temp_files):
        with h5py.File(temp_path, 'r') as temp_file:
            # 假设每个临时文件里有一个'data'数据集
            data = temp_file['data'][()]
            # 方式1：按部分存储
            final_file.create_dataset(f'data_part_{idx}', data=data)
            # 方式2：合并到同一个数据集
            # if idx == 0:
            #     dset = final_file.create_dataset('merged_data', shape=(len(temp_files), len(data)), dtype=data.dtype)
            # dset[idx, :] = data

注意事项 #

• 不要尝试手动绕过HDF5的文件锁（比如修改文件权限或者强制删除锁文件），这会导致HDF5文件损坏，数据丢失。
• 如果是多进程读+单进程写的场景，可以考虑HDF5的SWMR（Single Writer Multiple Readers）模式，但该模式不支持多进程写入。

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