使用tf.data.Dataset.interleave并行读文件遇错，求解决方案

核心问题分析 #

1. 第一个错误：interleave传入的文件名是tf.Tensor类型，而你的read_binary_file期望接收常规Python字符串，导致TypeError。
2. 第二个错误：在生成器中直接迭代tf.py_function的返回值——图模式下符号tf.Tensor不允许被迭代，触发OperatorNotAllowedInGraphError。

正确实现方式 #

要实现并行读取，需要将单文件处理逻辑包装为返回tf.data.Dataset的函数，而非生成器，这样interleave可以正确处理并行生成的数据集。

步骤1：定义单文件处理函数

该函数接收文件名Tensor，通过tf.py_function调用自定义读取逻辑，再将读取结果转换为单个数据点组成的Dataset：

def _process_single_file(fname):
    # 内部函数：接收Python字符串文件名，返回所有数据点列表
    def py_read_file(fname_str):
        return read_binary_file(fname_str)
    
    # 用tf.py_function桥接Python逻辑与TensorFlow图
    data_points_tensor = tf.py_function(
        func=py_read_file,
        inp=[fname],
        Tout=tf.string
    )
    # 将包含所有数据点的Tensor拆分为单个元素的Dataset
    return tf.data.Dataset.from_tensor_slices(data_points_tensor)

步骤2：用interleave构建并行数据集

# 1. 创建文件名数据集
my_dataset = tf.data.Dataset.list_files(fnames)

# 2. 并行处理文件，cycle_length控制并行数
my_dataset = my_dataset.interleave(
    map_func=lambda fname: _process_single_file(fname),
    cycle_length=8,
    num_parallel_calls=tf.data.AUTOTUNE  # 自动适配并行数，提升效率
)

模拟read_binary_file示例（供参考）

假设你的二进制文件中每个数据点是固定长度（比如16字节），读取逻辑可以是：

def read_binary_file(fname):
    with open(fname, 'rb') as f:
        content = f.read()
    # 按固定长度拆分二进制内容为多个数据点
    point_size = 16
    data_points = [content[i:i+point_size] for i in range(0, len(content), point_size)]
    return data_points

为什么这样能解决问题？ #

• tf.py_function会自动将输入的Tensor转换为Python原生类型（这里是字符串文件名），传给你的read_binary_file函数。
• 读取得到的多数据点列表会被包装为一个字符串Tensor，再通过from_tensor_slices拆分为单个数据点的Dataset，避免了在图模式下迭代Tensor的操作。
• interleave会并行调用_process_single_file处理多个文件，实现你需要的并行读取效果。

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