深入 Ray Data 自定义数据源从零实现 Datasource 与 Datasink 读写任意文件格式【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray本指南围绕 Ray Data 的扩展点展开当内置的 read/write API 无法覆盖你所需的文件格式时如何通过继承FileBasedDatasource与RowBasedFileDatasink或BlockBasedFileDatasink实现自定义数据源与数据汇从而用ray.data.read_datasource/Dataset.write_datasink读写任意格式的文件。读完本文你将掌握自定义数据源从构造函数、_read_stream到并行读取的完整链路以及按行/按块写文件的两种 datasink 基类用法并了解 Ray Data 内部真实图片数据源 image_datasource.py 如何落实这些接口。提示本文属于进阶指南需要接触 Ray Data 中尚未稳定的内部 API如DelegatingBlockBuilder、Block 抽象相关接口在源码中标记为DeveloperAPI参见 datasource.py 与 datasink.py接口签名可能随版本演进。先判断你真的需要自定义 Datasource 吗Ray Data 已经内置了对图片、CSV、JSON、Parquet、文本、二进制等多种格式的读取支持。本文以图片为例讲解如何实现自定义数据源只是为了教学演示——官方建议若非必须直接用ray.data.read_images与Dataset.write_images即可。一个更轻量级的替代方案是不创建Datasource子类而是调用ray.data.read_binary_files读取任意文件的原始字节再通过Dataset.map在分布式任务中解码。例如读取图片可以先read_binary_files拿到bytes列再用 PIL 在map里解码。只有当这种方式无法满足需求例如需要高度优化的并行解析、文件级元数据、扩展名过滤等才值得实现自定义 Datasource。读文件的核心抽象FileBasedDatasource读取文件的核心抽象是FileBasedDatasource定义于 file_based_datasource.py它在底层Datasource接口之上封装了文件系统相关的通用能力路径解析、文件扩展名过滤、文件大小获取、read task 切分与并行调度、流式打开文件含压缩推断、S3 序列化兼容处理等。子类只需要做两件事实现构造函数调用父类构造函数并声明要读取的文件实现_read_stream把单个文件流解析为一个或多个 Block。第一步实现构造函数构造函数中调用父类构造函数并传入要读取的路径可选地通过file_extensions声明合法文件扩展名Ray Data 会过滤掉其他扩展名的文件若过滤后无文件会直接抛出ValueError见 file_based_datasource.py。除此之外还可以在此构造器中保存自己的读取选项如示例中的mode。from ray.data.datasource import FileBasedDatasource class ImageDatasource(FileBasedDatasource): def __init__(self, paths: Union[str, List[str]], *, mode: str): super().__init__( paths, file_extensions[png, jpg, jpeg, bmp, gif, tiff], ) self.mode mode # Specify read options in the constructor父类构造函数FileBasedDatasource.__init__支持以下常用参数可用于更复杂的自定义数据源filesystemPyArrow 文件系统实现不传则根据路径推断本地、S3 等schema可选的显式 schemaopen_stream_args打开输入流时透传给open_input_stream的参数meta_provider文件元数据提供器如文件大小、行数估计partition_filter/partitioning目录分区过滤与分区信息解析ignore_missing_paths是否忽略不存在的路径shuffle文件级打乱策略None、files或FileShuffleConfiginclude_paths是否把文件路径作为path列加入结果file_extensions合法扩展名白名单。第二步实现_read_stream_read_stream是一个生成器接收已打开的文件流fpyarrow.NativeFile与文件路径path逐个 yield 数据 Block。Block 是 Ray Data 内部的行集合抽象可以是 PyArrow Table、pandas DataFrame 或 NumPy 数组字典见 block.py。不要直接手工构造 Block而应把行数据逐条加入DelegatingBlockBuilder定义于 delegating_block_builder.py由它根据数据类型自动选择合适的 Block 表示最后yield builder.build()。def _read_stream(self, f: pyarrow.NativeFile, path: str) - Iterator[Block]: import io import numpy as np from PIL import Image from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder data f.readall() image Image.open(io.BytesIO(data)) image image.convert(self.mode) # Each block contains one row builder DelegatingBlockBuilder() array np.asarray(image) item {image: array} builder.add(item) yield builder.build()该示例每个文件产出一个包含单行的 Block行是一个{image: np.ndarray}字典。_read_stream也可以 yield 多个 Block例如超大文件分块产出避免内存峰值Ray Data 的ReadTask机制允许单任务返回多个 Block见 datasource.py 中对ReadTask的说明。第三步用 read_datasource 并行读取实现完ImageDatasource后通过ray.data.read_datasource把它接入 Dataset 读取流程。Ray Data 会自动把文件列表切成多个 read task 并行执行——从源码看FileBasedDatasource.get_read_tasks会根据parallelism对路径做np.array_split切分并为每个分片生成一个ReadTask见 file_based_datasource.py每个 task 在 Ray worker 上远程执行即文件级并行。import ray ds ray.data.read_datasource( ImageDatasource(s3://anonymousray-example-data/batoidea, modeRGB) )read_datasource定义于 read_api.py还支持通过num_cpus、num_gpus、memory、concurrency、computeTaskPoolStrategy/ActorPoolStrategy、override_num_blocks、label_selector、ray_remote_args等参数精细化控制读取任务的计算资源与调度策略。写文件的核心抽象RowBasedFileDatasink 与 BlockBasedFileDatasink写数据侧的核心抽象是RowBasedFileDatasink与BlockBasedFileDatasink均定义于 file_datasink.py它们在Datasink接口之上封装了目标目录创建、SaveMode 冲突处理、文件名生成、写流重试等通用逻辑RowBasedFileDatasink每行数据写一个文件适合图片、单条记录等一行一文件的场景BlockBasedFileDatasink每个数据块写一个文件适合 CSV 等一块一文件、需要批量写入的场景。示例中要写每张图片一个文件因此继承RowBasedFileDatasink需实现构造函数与write_row_to_file。第一步实现构造函数调用父类构造函数并指定写入目录可选地传入file_format字符串如pngRay Data 会把它作为文件扩展名追加到生成的每个文件名上。from ray.data.datasource import RowBasedFileDatasink class ImageDatasink(RowBasedFileDatasink): def __init__(self, path: str, column: str, file_format: str): super().__init__(path, file_formatfile_format) self.column column self.file_format file_format # Specify write options in the constructor父类_FileDatasink还支持filesystem、try_create_dir、open_stream_args、filename_provider、dataset_uuid、mode等参数。其中mode接受SaveMode枚举值源码 file_datasink.py 中的on_write_start展示了其行为目标已存在时ERROR/CREATE抛错、IGNORE跳过写入、OVERWRITE清空目录后重写、APPEND默认直接追加写入期间若目录不存在会自动递归创建。第二步实现 write_row_to_filewrite_row_to_file负责把一行数据写进一个文件流。每个row是列名到值的字典。由于 PIL 不能直接写pyarrow.NativeFile示例先写进内存BytesIO缓冲区再把字节一次性写入文件流def write_row_to_file(self, row: Dict[str, Any], file: pyarrow.NativeFile): import io from PIL import Image # PIL cant write to a NativeFile, so we have to write to a buffer first. image Image.fromarray(row[self.column]) buffer io.BytesIO() image.save(buffer, formatself.file_format) file.write(buffer.getvalue())从源码 file_datasink.py 可以看到RowBasedFileDatasink.write_block会遍历块中的每一行为每行生成{task_base}_{block_index:06}_{row_index:06}{ext}形式的文件名并通过call_with_retry包一层 IO 错误重试后调用你的write_row_to_file。如果希望每行写一个文件只需实现这一个方法文件名与目录管理完全交给基类。第三步用 write_datasink 并行写文件实现完ImageDatasink后在任意 Dataset 上调用write_datasink见 dataset.pyRay Data 会按块并行执行写任务ds.write_datasink(ImageDatasink(/tmp/results, columnimage, file_formatpng))若数据分布在多个 BlockRay Data 会并行写入多个文件无需手动管理并发。若你的格式更适合一块一文件改继承BlockBasedFileDatasink并实现write_block_to_file(block, file)即可基类还支持min_rows_per_file参数控制每个文件的目标最少行数见 file_datasink.py。对照源码Ray Data 内置图片数据源是怎么写的python/ray/data/_internal/datasource/image_datasource.py中内置的ImageDatasource与本例结构完全一致可作为进阶参考它展示了生产级实现中额外的细节类属性_FILE_EXTENSIONS [png, jpg, jpeg, tif, tiff, bmp, gif]并在类级别声明供read_images的file_extensions参数默认引用_NUM_THREADS_PER_TASK 8每个 read task 内部用 8 个线程并发读取多个文件FileBasedDatasource会据此用make_async_gen组织并发见 file_based_datasource.pysize参数支持统一缩放Image.resizemode参数支持颜色模式转换convert并校验非法参数estimate_inmemory_data_size结合文件大小与编码比率估算内存占用供调度器决策。写侧对应python/ray/data/_internal/datasource/image_datasink.py中的ImageDatasink同样是RowBasedFileDatasink子类实现方式与本文示例如出一辙。read_images/write_images的完整签名可在 read_api.py 与 dataset.py 中查看内置实现天然支持include_paths、partitioning目录分区与shuffle等能力——如果你的自定义数据源也需要这些可继承FileBasedDatasource直接复用partitioning、include_paths的处理逻辑已在 file_based_datasource.py 中实现。完整示例与注意事项本文所有代码段均来自仓库中的完整可运行示例 doc/source/data/doc_code/custom_datasource_example.py其中还包含typing、pyarrow、Block等前置导入。使用前请确认环境已安装pillow、numpy、pyarrow。需要注意的几点接口稳定性写接口Datasink仍在活跃开发中未来可能变化如你有功能诉求可通过 Ray 官方 GitHub Issue 反馈仓库文档中标注了此提示。内部 APIDelegatingBlockBuilder位于ray.data._internal属于内部实现跨版本可能调整。本地文件与 Ray Client源码 file_based_datasource.py 表明本地路径的读取任务无法在 Ray Client 连接模式下访问驱动节点的本地文件请改用云存储或 NFS 等分布式文件系统。并行度FileBasedDatasource会把parallelism限制在文件数量以内parallelism min(parallelism, len(paths))即并行上限是文件数单文件内部的并行由_NUM_THREADS_PER_TASK控制。掌握了FileBasedDatasourceRowBasedFileDatasink/BlockBasedFileDatasink这套扩展点后无论是自研的二进制协议、专有图像格式还是任意自定义文本格式都能以几十行代码接入 Ray Data 的分布式读写管线。【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 深入 Ray Data 自定义数据源：从零实现 Datasource 与 Datasink 读写任意文件格式
深入 Ray Data 自定义数据源从零实现 Datasource 与 Datasink 读写任意文件格式【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray本指南围绕 Ray Data 的扩展点展开当内置的 read/write API 无法覆盖你所需的文件格式时如何通过继承FileBasedDatasource与RowBasedFileDatasink或BlockBasedFileDatasink实现自定义数据源与数据汇从而用ray.data.read_datasource/Dataset.write_datasink读写任意格式的文件。读完本文你将掌握自定义数据源从构造函数、_read_stream到并行读取的完整链路以及按行/按块写文件的两种 datasink 基类用法并了解 Ray Data 内部真实图片数据源 image_datasource.py 如何落实这些接口。提示本文属于进阶指南需要接触 Ray Data 中尚未稳定的内部 API如DelegatingBlockBuilder、Block 抽象相关接口在源码中标记为DeveloperAPI参见 datasource.py 与 datasink.py接口签名可能随版本演进。先判断你真的需要自定义 Datasource 吗Ray Data 已经内置了对图片、CSV、JSON、Parquet、文本、二进制等多种格式的读取支持。本文以图片为例讲解如何实现自定义数据源只是为了教学演示——官方建议若非必须直接用ray.data.read_images与Dataset.write_images即可。一个更轻量级的替代方案是不创建Datasource子类而是调用ray.data.read_binary_files读取任意文件的原始字节再通过Dataset.map在分布式任务中解码。例如读取图片可以先read_binary_files拿到bytes列再用 PIL 在map里解码。只有当这种方式无法满足需求例如需要高度优化的并行解析、文件级元数据、扩展名过滤等才值得实现自定义 Datasource。读文件的核心抽象FileBasedDatasource读取文件的核心抽象是FileBasedDatasource定义于 file_based_datasource.py它在底层Datasource接口之上封装了文件系统相关的通用能力路径解析、文件扩展名过滤、文件大小获取、read task 切分与并行调度、流式打开文件含压缩推断、S3 序列化兼容处理等。子类只需要做两件事实现构造函数调用父类构造函数并声明要读取的文件实现_read_stream把单个文件流解析为一个或多个 Block。第一步实现构造函数构造函数中调用父类构造函数并传入要读取的路径可选地通过file_extensions声明合法文件扩展名Ray Data 会过滤掉其他扩展名的文件若过滤后无文件会直接抛出ValueError见 file_based_datasource.py。除此之外还可以在此构造器中保存自己的读取选项如示例中的mode。from ray.data.datasource import FileBasedDatasource class ImageDatasource(FileBasedDatasource): def __init__(self, paths: Union[str, List[str]], *, mode: str): super().__init__( paths, file_extensions[png, jpg, jpeg, bmp, gif, tiff], ) self.mode mode # Specify read options in the constructor父类构造函数FileBasedDatasource.__init__支持以下常用参数可用于更复杂的自定义数据源filesystemPyArrow 文件系统实现不传则根据路径推断本地、S3 等schema可选的显式 schemaopen_stream_args打开输入流时透传给open_input_stream的参数meta_provider文件元数据提供器如文件大小、行数估计partition_filter/partitioning目录分区过滤与分区信息解析ignore_missing_paths是否忽略不存在的路径shuffle文件级打乱策略None、files或FileShuffleConfiginclude_paths是否把文件路径作为path列加入结果file_extensions合法扩展名白名单。第二步实现_read_stream_read_stream是一个生成器接收已打开的文件流fpyarrow.NativeFile与文件路径path逐个 yield 数据 Block。Block 是 Ray Data 内部的行集合抽象可以是 PyArrow Table、pandas DataFrame 或 NumPy 数组字典见 block.py。不要直接手工构造 Block而应把行数据逐条加入DelegatingBlockBuilder定义于 delegating_block_builder.py由它根据数据类型自动选择合适的 Block 表示最后yield builder.build()。def _read_stream(self, f: pyarrow.NativeFile, path: str) - Iterator[Block]: import io import numpy as np from PIL import Image from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder data f.readall() image Image.open(io.BytesIO(data)) image image.convert(self.mode) # Each block contains one row builder DelegatingBlockBuilder() array np.asarray(image) item {image: array} builder.add(item) yield builder.build()该示例每个文件产出一个包含单行的 Block行是一个{image: np.ndarray}字典。_read_stream也可以 yield 多个 Block例如超大文件分块产出避免内存峰值Ray Data 的ReadTask机制允许单任务返回多个 Block见 datasource.py 中对ReadTask的说明。第三步用 read_datasource 并行读取实现完ImageDatasource后通过ray.data.read_datasource把它接入 Dataset 读取流程。Ray Data 会自动把文件列表切成多个 read task 并行执行——从源码看FileBasedDatasource.get_read_tasks会根据parallelism对路径做np.array_split切分并为每个分片生成一个ReadTask见 file_based_datasource.py每个 task 在 Ray worker 上远程执行即文件级并行。import ray ds ray.data.read_datasource( ImageDatasource(s3://anonymousray-example-data/batoidea, modeRGB) )read_datasource定义于 read_api.py还支持通过num_cpus、num_gpus、memory、concurrency、computeTaskPoolStrategy/ActorPoolStrategy、override_num_blocks、label_selector、ray_remote_args等参数精细化控制读取任务的计算资源与调度策略。写文件的核心抽象RowBasedFileDatasink 与 BlockBasedFileDatasink写数据侧的核心抽象是RowBasedFileDatasink与BlockBasedFileDatasink均定义于 file_datasink.py它们在Datasink接口之上封装了目标目录创建、SaveMode 冲突处理、文件名生成、写流重试等通用逻辑RowBasedFileDatasink每行数据写一个文件适合图片、单条记录等一行一文件的场景BlockBasedFileDatasink每个数据块写一个文件适合 CSV 等一块一文件、需要批量写入的场景。示例中要写每张图片一个文件因此继承RowBasedFileDatasink需实现构造函数与write_row_to_file。第一步实现构造函数调用父类构造函数并指定写入目录可选地传入file_format字符串如pngRay Data 会把它作为文件扩展名追加到生成的每个文件名上。from ray.data.datasource import RowBasedFileDatasink class ImageDatasink(RowBasedFileDatasink): def __init__(self, path: str, column: str, file_format: str): super().__init__(path, file_formatfile_format) self.column column self.file_format file_format # Specify write options in the constructor父类_FileDatasink还支持filesystem、try_create_dir、open_stream_args、filename_provider、dataset_uuid、mode等参数。其中mode接受SaveMode枚举值源码 file_datasink.py 中的on_write_start展示了其行为目标已存在时ERROR/CREATE抛错、IGNORE跳过写入、OVERWRITE清空目录后重写、APPEND默认直接追加写入期间若目录不存在会自动递归创建。第二步实现 write_row_to_filewrite_row_to_file负责把一行数据写进一个文件流。每个row是列名到值的字典。由于 PIL 不能直接写pyarrow.NativeFile示例先写进内存BytesIO缓冲区再把字节一次性写入文件流def write_row_to_file(self, row: Dict[str, Any], file: pyarrow.NativeFile): import io from PIL import Image # PIL cant write to a NativeFile, so we have to write to a buffer first. image Image.fromarray(row[self.column]) buffer io.BytesIO() image.save(buffer, formatself.file_format) file.write(buffer.getvalue())从源码 file_datasink.py 可以看到RowBasedFileDatasink.write_block会遍历块中的每一行为每行生成{task_base}_{block_index:06}_{row_index:06}{ext}形式的文件名并通过call_with_retry包一层 IO 错误重试后调用你的write_row_to_file。如果希望每行写一个文件只需实现这一个方法文件名与目录管理完全交给基类。第三步用 write_datasink 并行写文件实现完ImageDatasink后在任意 Dataset 上调用write_datasink见 dataset.pyRay Data 会按块并行执行写任务ds.write_datasink(ImageDatasink(/tmp/results, columnimage, file_formatpng))若数据分布在多个 BlockRay Data 会并行写入多个文件无需手动管理并发。若你的格式更适合一块一文件改继承BlockBasedFileDatasink并实现write_block_to_file(block, file)即可基类还支持min_rows_per_file参数控制每个文件的目标最少行数见 file_datasink.py。对照源码Ray Data 内置图片数据源是怎么写的python/ray/data/_internal/datasource/image_datasource.py中内置的ImageDatasource与本例结构完全一致可作为进阶参考它展示了生产级实现中额外的细节类属性_FILE_EXTENSIONS [png, jpg, jpeg, tif, tiff, bmp, gif]并在类级别声明供read_images的file_extensions参数默认引用_NUM_THREADS_PER_TASK 8每个 read task 内部用 8 个线程并发读取多个文件FileBasedDatasource会据此用make_async_gen组织并发见 file_based_datasource.pysize参数支持统一缩放Image.resizemode参数支持颜色模式转换convert并校验非法参数estimate_inmemory_data_size结合文件大小与编码比率估算内存占用供调度器决策。写侧对应python/ray/data/_internal/datasource/image_datasink.py中的ImageDatasink同样是RowBasedFileDatasink子类实现方式与本文示例如出一辙。read_images/write_images的完整签名可在 read_api.py 与 dataset.py 中查看内置实现天然支持include_paths、partitioning目录分区与shuffle等能力——如果你的自定义数据源也需要这些可继承FileBasedDatasource直接复用partitioning、include_paths的处理逻辑已在 file_based_datasource.py 中实现。完整示例与注意事项本文所有代码段均来自仓库中的完整可运行示例 doc/source/data/doc_code/custom_datasource_example.py其中还包含typing、pyarrow、Block等前置导入。使用前请确认环境已安装pillow、numpy、pyarrow。需要注意的几点接口稳定性写接口Datasink仍在活跃开发中未来可能变化如你有功能诉求可通过 Ray 官方 GitHub Issue 反馈仓库文档中标注了此提示。内部 APIDelegatingBlockBuilder位于ray.data._internal属于内部实现跨版本可能调整。本地文件与 Ray Client源码 file_based_datasource.py 表明本地路径的读取任务无法在 Ray Client 连接模式下访问驱动节点的本地文件请改用云存储或 NFS 等分布式文件系统。并行度FileBasedDatasource会把parallelism限制在文件数量以内parallelism min(parallelism, len(paths))即并行上限是文件数单文件内部的并行由_NUM_THREADS_PER_TASK控制。掌握了FileBasedDatasourceRowBasedFileDatasink/BlockBasedFileDatasink这套扩展点后无论是自研的二进制协议、专有图像格式还是任意自定义文本格式都能以几十行代码接入 Ray Data 的分布式读写管线。【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考