Databricks集群中Worker节点直接读写HDF文件的问题

错误原因 #

Worker节点通过FUSE挂载的/dbfs路径不支持HDF5写入所需的随机访问和文件锁机制。DBFS本质是基于对象存储（如S3、ADLS）的抽象层，对象存储本身不支持本地文件系统的随机修改操作，而HDF5在创建数据集时需要频繁的随机写入与锁操作，这会触发Operation not supported错误。Driver节点能成功是因为其/dbfs挂载的权限或访问方式与Worker不同，支持这类操作。

解决方案 #

方案1：先写Worker本地临时目录，再复制到DBFS #

这是最稳妥的兼容方案，利用Worker本地磁盘支持完整POSIX特性的优势，写完后再同步到DBFS路径。

修改后的代码：

def create_hdf_file_tmp(x):
    import numpy as np
    import h5py, os
    import pandas as pd
    from pyspark.dbutils import DBUtils
    import pyspark.sql

    # 初始化dbutils
    spark = pyspark.sql.SparkSession.builder.getOrCreate()
    dbutils = DBUtils(spark)

    # 生成唯一本地临时文件路径（避免多任务冲突）
    local_tmp_path = f"/tmp/demo_{os.getpid()}.hdf"
    dummy_data = [1,2,3,4,5]
    df_data = pd.DataFrame(dummy_data, columns=['Numbers'])

    # 写入Worker本地临时文件
    with h5py.File(local_tmp_path, 'w') as f:
        dset = f.create_dataset('default', data = df_data)
    
    # 复制到目标DBFS路径
    dbutils.fs.cp(f"file:{local_tmp_path}", x)
    
    # 清理本地临时文件
    os.remove(local_tmp_path)
    
    return True

# Driver代码不变
rdd = spark.sparkContext.parallelize(['/dbfs/mnt/demo.hdf'])
result = rdd.map(lambda x: create_hdf_file_tmp(x)).collect()

方案2：直接使用对象存储路径（如S3/ADLS）绕过FUSE挂载 #

如果你的/dbfs/mnt挂载的是云对象存储（如AWS S3、Azure ADLS），可以用h5py配合对应对象存储的文件系统后端，直接通过对象存储URL写入，绕过DBFS的FUSE限制。

步骤1：安装依赖

在Databricks notebook中执行：

%pip install h5py s3fs  # S3用s3fs，ADLS用adlfs

步骤2：修改代码（以S3为例）

def create_hdf_file_tmp(x):
    import numpy as np
    import h5py
    import pandas as pd
    import s3fs

    # 将/dbfs/mnt路径转换为S3原生路径（需根据实际挂载配置调整）
    # 例：/dbfs/mnt/demo.hdf 对应 s3://my-bucket/path/demo.hdf
    s3_path = x.replace("/dbfs/mnt/", "s3://your-bucket-name/")
    dummy_data = [1,2,3,4,5]
    df_data = pd.DataFrame(dummy_data, columns=['Numbers'])

    # 用s3fs作为后端写入HDF5
    fs = s3fs.S3FileSystem()
    with fs.open(s3_path, 'wb') as f:
        with h5py.File(f, 'w') as hdf_file:
            dset = hdf_file.create_dataset('default', data = df_data)
    
    return True

# Driver代码不变
rdd = spark.sparkContext.parallelize(['/dbfs/mnt/demo.hdf'])
result = rdd.map(lambda x: create_hdf_file_tmp(x)).collect()

注意事项 #

• 方案2需要确保Worker节点有对应对象存储的访问权限（通过IAM角色、密钥等配置）。
• 避免多个Worker任务同时写入同一个HDF文件，会导致数据损坏，建议每个任务写入独立文件后再合并。

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