Ray Data 数据检查完整指南schema、行/批次采样与执行统计【免费下载链接】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 把数据集建模为带 schema 的表格结构并提供了一组轻量 APIschema()、take()、take_batch()、stats()让你在不触发全量执行的前提下快速掌握列名与类型、行数、样本行、批次形态以及每个算子的执行耗时与内存占用。本文基于 inspecting-data.rst 编写并结合 dataset.py 等源码给出底层实现细节读完后你将能像调试普通表格一样调试任意规模的 Ray Data 数据集。概览检查数据能解决什么问题Ray Data 的Dataset是惰性lazy的read_csv()、map_batches()等调用只是构建逻辑执行计划并不会立即搬数据。因此在投入昂贵的全量计算之前先做以下四类检查是最经济的调试手段描述数据集schema查看列名、列类型、行数确认数据形态符合预期检查行take取出少量行Python dict做内容抽查检查批次take_batch按下游算子尤其是map_batches真正消费的形态NumPy / pandas / PyArrow抽查数据检查执行统计stats了解每个算子的耗时、吞吐、内存与任务分布定位性能瓶颈。下文逐一展开每个小节都附可直接运行的代码与可验证的源码依据。描述数据集schema() 与打印 DatasetDataset是表格化的查看列名与列类型直接调用 Dataset.schema()import ray ds ray.data.read_csv(s3://anonymousair-example-data/iris.csv) print(ds.schema())输出Column Type ------ ---- sepal length (cm) double sepal width (cm) double petal length (cm) double petal width (cm) double target int64对 Iris 数据集五列分别是四个浮点型花萼/花瓣尺寸与一个整型目标类别。如果想连行数一起看直接print(ds)print(ds) # Dataset(num_rows..., schema...)num_rows会随数据源不同而给出确定值例如 Parquet 可直接从文件元数据拿到行数schema部分展示与schema()相同的列信息。schema() 的源码行为在 dataset.py 中schema(fetch_if_missingTrue)的时间复杂度是 O(1)。其内部通过_base_schema()优先从缓存读取若 schema 未知且fetch_if_missingTrue它会惰性执行limit(1)只读取第一个 block 来推断 schema而不是扫描整个数据集见 dataset.py 的_base_schema实现。如果你明确不想触发任何执行可以传schema(fetch_if_missingFalse)此时未知 schema 返回None——这正是 test_consumption.py 中test_schema用fetch_if_missingFalse断言不产生任何执行任务的原因。此外还有两个高频伴随 APIds.columns()仅返回列名列表如[sepal length (cm), ..., target]同样支持fetch_if_missing参数dataset.pyds.count()返回总行数。对仅由read_parquet创建的 Datasetcount()直接读取 Parquet 元数据计数不读取具体数据非常高效对普通数据集则通过一个轻量的Count逻辑算子统计dataset.py。检查行take() 与 take_all()要拿少量行做内容抽查使用 Dataset.take() 或 Dataset.take_all()。Ray Data 将每一行表示为一个 Python字典import ray ds ray.data.read_csv(s3://anonymousair-example-data/iris.csv) rows ds.take(1) print(rows)输出[{sepal length (cm): 5.1, sepal width (cm): 3.5, petal length (cm): 1.4, petal width (cm): 0.2, target: 0}]两个方法的关键差异与使用要点方法返回默认参数适用场景风险take(limit20)最多limit行组成的 listlimit20快速抽查任意数据集limit过大时数据被拉回调用方机器可能 OOMtake_all(limitNone)全部行组成的 list不设上限只适用于小数据集会把整个数据集拉到调用方机器大数据集必 OOMshow(limit20)无逐行 printlimit20终端里直接看数据同上take_all(limit...)还有一个保护语义如果数据行数超过给定limit会直接抛出ValueError防止你误对大数据集调用。源码中take与take_all都基于iter_rows()逐行产出并组装dataset.py、dataset.py且take内部先执行limit(limit)再迭代因此时间复杂度是 O(limit)。值得注意take()在首次被调用时会打印一条提示日志建议优先用take_batch()以 pandas / numpy 批次格式取数见 dataset.py 的log_once分支。测试 test_take_all 也验证了take_all(4)对 5 行数据集抛出ValueError的行为。拿到行之后可以进一步做行级变换如map、flat_map或逐行迭代参见 Transforming rows 与 Iterating over rows。检查批次take_batch() 与 batch_format行视图适合人工浏览但 Ray Data 的map_batches等算子实际消费的是批次batch——一个 batch 包含多行数据。用 Dataset.take_batch() 可以按下游真正面对的形态来检查数据import ray # 图片数据集默认 batch_format 为 numpy得到 dict[str, np.ndarray] ds ray.data.read_images(s3://anonymousray-example-data/image-datasets/simple) batch ds.take_batch(batch_size2, batch_formatnumpy) print(Batch:, batch) print(Image shape, batch[image].shape)输出示意Batch: {image: array([[[[...]]]], dtypeuint8)} Image shape: (2, 32, 32, 3)batch_format可选值由block.py中的VALID_BATCH_FORMATS定义block.pybatch_format返回类型说明default/numpyDict[str, numpy.ndarray]默认值列名映射到 NumPy 数组pandaspandas.DataFrame适合与 pandas 生态代码衔接pyarrowpyarrow.Table零拷贝、列式存储cudfcudf.DataFrameGPU 加速实验特性一个容易被忽略的关键点batch_format 只决定返回给调用方的表示形式与 Ray Data 底层 block 的存储格式完全无关。也就是说无论内部 block 是 Arrow 还是其他格式你都可以按需指定任一种 batch_format 取数无需关心内部实现。pandas 示例import ray ds ray.data.read_csv(s3://anonymousair-example-data/iris.csv) batch ds.take_batch(batch_size2, batch_formatpandas) print(batch)输出sepal length (cm) sepal width (cm) ... petal width (cm) target 0 5.1 3.5 ... 0.2 0 1 4.9 3.0 ... 0.2 0pyarrow 示例import ray ds ray.data.read_csv(s3://anonymousair-example-data/iris.csv) batch ds.take_batch(batch_size2, batch_formatpyarrow) print(batch)输出pyarrow.Table sepal length (cm): double sepal width (cm): double petal length (cm): double petal width (cm): double target: int64 ---- sepal length (cm): [[5.1,4.9]] sepal width (cm): [[3.5,3]] petal length (cm): [[1.4,1.4]] petal width (cm): [[0.2,0.2]] target: [[0,0]]take_batch() 的源码实现与边界行为take_batch(batch_size20, batch_formatdefault)的实现非常直白dataset.py调用_apply_batch_format()把default解析为DEFAULT_BATCH_FORMAT numpy并校验格式合法性非法值抛出ValueErrorblock.py对数据集先做limit(batch_size)以prefetch_batches0、指定batch_format调用iter_batches()并取第一个批次若数据集为空StopIteration抛出ValueError(The dataset is empty.)。因此它同样有最多返回batch_size行到调用方机器的内存警示batch_size 过大时调用方可能 OOM。测试 test_take_batch 验证了take_batch(3)返回前 3 行、batch_size超过总行数时返回全部行、pandas格式返回pd.DataFrame、numpy 格式返回dict以及空数据集抛ValueError等全部边界。需要注意的是read_images()的 schema图片列默认列名为image类型为ArrowTensorTypeV2(shape(32, 32, 3), dtypeuint8)见 read_api.py所以上面batch[image].shape得到(2, 32, 32, 3)正好是batch 大小 × 高 × 宽 × 通道。拿到批次后若想深入理解批级变换与迭代参见 Transforming batches 与 Iterating over batches。检查执行统计stats() 与日志落盘Ray Data 在执行期间为每个算子统计指标包括墙钟时间wall clock time、CPU 时间、block 变换耗时、峰值堆内存、输出行数/字节数、任务分布与算子吞吐等。对已执行的数据集调用 Dataset.stats() 即可查看import ray from huggingface_hub import HfFileSystem def f(batch): return batch def g(row): return True path hf://datasets/ylecun/mnist/mnist/ fs HfFileSystem() train_files [f[name] for f in fs.ls(path) if train in f[name] and f[name].endswith(.parquet)] ds ( ray.data.read_parquet(train_files, filesystemfs) .map_batches(f) .filter(g) .materialize() ) print(ds.stats())输出示意数值随机器与版本浮动Operator 1 ReadParquet-SplitBlocks(32): 1 tasks executed, 32 blocks produced in 2.92s * Remote wall time: 103.38us min, 1.34s max, 42.14ms mean, 1.35s total * Remote cpu time: 102.0us min, 164.66ms max, 5.37ms mean, 171.72ms total * Block transform time: 95.12us min, 1.31s max, 41.09ms mean, 1.31s total * Peak heap memory usage (MiB): 266375.0 min, 281875.0 max, 274491 mean * Output num rows per block: 1875 min, 1875 max, 1875 mean, 60000 total * Output size bytes per block: 537986 min, 555360 max, 545963 mean, 17470820 total * Output rows per task: 60000 min, 60000 max, 60000 mean, 1 tasks used * Tasks per node: 1 min, 1 max, 1 mean; 1 nodes used * Operator throughput: * Ray Data throughput: 20579.80984833993 rows/s * Estimated single node throughput: 44492.67361278733 rows/s Operator 2 MapBatches(f)-Filter(g): 32 tasks executed, 32 blocks produced in 3.63s * Remote wall time: 675.48ms min, 1.0s max, 797.07ms mean, 25.51s total * Remote cpu time: 673.41ms min, 897.32ms max, 768.09ms mean, 24.58s total * Block transform time: 661.65ms min, 978.04ms max, 778.13ms mean, 24.9s total * Peak heap memory usage (MiB): 152281.25 min, 286796.88 max, 164231 mean * Output num rows per block: 1875 min, 1875 max, 1875 mean, 60000 total * Output size bytes per block: 530251 min, 547625 max, 538228 mean, 17223300 total * Output rows per task: 1875 min, 1875 max, 1875 mean, 32 tasks used * Tasks per node: 32 min, 32 max, 32 mean; 1 nodes used * Operator throughput: * Ray Data throughput: 16512.364546087643 rows/s * Estimated single node throughput: 2352.3683708977856 rows/s Dataset throughput: * Ray Data throughput: 11463.372316361854 rows/s * Estimated single node throughput: 25580.963670075285 rows/s如何解读 stats() 输出对每个物理算子重点看以下几类指标时间指标Remote wall time任务真实墙钟时间、Remote cpu timeCPU 耗时、Block transform timeblock 内数据变换耗时。三者差距大通常意味着存在 IO 等待或调度开销内存指标Peak heap memory usage (MiB)给出 min/max/mean 三个值用于判断是否存在内存尖峰产出指标Output num rows per block、Output size bytes per block、Output rows per task帮助判断 block 划分是否均匀、是否需要调整并行度分布指标Tasks per node反映数据在节点间的分布吞吐指标Ray Data throughput是包含调度开销在内的端到端吞吐Estimated single node throughput是排除分布式开销后的估算值——两者差距大说明分布式调度/网络成为瓶颈可考虑提升任务粒度或数据本地性。两个重要的使用前提stats() 不会触发执行如果数据集尚未执行stats()返回空字符串。必须先通过materialize()、iter_batches()等操作真正执行管线再调用stats()查看源码注释与示例见 dataset.py统计信息会持久化到日志文件每个算子的 stats 同时以日志形式写入/tmp/ray/session_*/logs/ray-data/ray-data.log。该路径由 Ray Data 的日志配置决定——logging.py 中为ray.datalogger 注册了名为ray-data.log的SessionFileHandler落在当前 Ray session 目录下。更完整的观测手段stats()是文本形态的统计快照。若要持续观测Ray Data 还提供进度条、Ray DashboardRay Data Overview 表、Metrics 时间序列视图以及 Prometheus 指标data_output_rows、data_output_bytes、data_cpu_usage_cores、data_gpu_usage_cores等按 dataset/operator 标签区分详见 Monitoring Your Workload。总结与进一步阅读检查数据是 Ray Data 使用流程中成本最低、收益最高的环节ds.schema()O(1)惰性推断与print(ds)回答数据长什么样、有多少行ds.take()/ds.take_all()/ds.show()回答行的内容对不对ds.take_batch(batch_format...)回答下游批次算子的输入形态对不对且返回格式与内部存储解耦ds.stats()回答每个算子花在哪、吞吐与内存如何并落盘到ray-data.log。建议的实践路径读取数据后先schema()take(5)确认结构再在写map_batches前用take_batch(batch_formatpandas)验证批次形态最后对整条管线materialize()后查看stats()定位热点算子。相关主题还可继续阅读 Iterating over data、Transforming data 与 Key concepts。【免费下载链接】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 数据检查完整指南schema、行/批次采样与执行统计【免费下载链接】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 把数据集建模为带 schema 的表格结构并提供了一组轻量 APIschema()、take()、take_batch()、stats()让你在不触发全量执行的前提下快速掌握列名与类型、行数、样本行、批次形态以及每个算子的执行耗时与内存占用。本文基于 inspecting-data.rst 编写并结合 dataset.py 等源码给出底层实现细节读完后你将能像调试普通表格一样调试任意规模的 Ray Data 数据集。概览检查数据能解决什么问题Ray Data 的Dataset是惰性lazy的read_csv()、map_batches()等调用只是构建逻辑执行计划并不会立即搬数据。因此在投入昂贵的全量计算之前先做以下四类检查是最经济的调试手段描述数据集schema查看列名、列类型、行数确认数据形态符合预期检查行take取出少量行Python dict做内容抽查检查批次take_batch按下游算子尤其是map_batches真正消费的形态NumPy / pandas / PyArrow抽查数据检查执行统计stats了解每个算子的耗时、吞吐、内存与任务分布定位性能瓶颈。下文逐一展开每个小节都附可直接运行的代码与可验证的源码依据。描述数据集schema() 与打印 DatasetDataset是表格化的查看列名与列类型直接调用 Dataset.schema()import ray ds ray.data.read_csv(s3://anonymousair-example-data/iris.csv) print(ds.schema())输出Column Type ------ ---- sepal length (cm) double sepal width (cm) double petal length (cm) double petal width (cm) double target int64对 Iris 数据集五列分别是四个浮点型花萼/花瓣尺寸与一个整型目标类别。如果想连行数一起看直接print(ds)print(ds) # Dataset(num_rows..., schema...)num_rows会随数据源不同而给出确定值例如 Parquet 可直接从文件元数据拿到行数schema部分展示与schema()相同的列信息。schema() 的源码行为在 dataset.py 中schema(fetch_if_missingTrue)的时间复杂度是 O(1)。其内部通过_base_schema()优先从缓存读取若 schema 未知且fetch_if_missingTrue它会惰性执行limit(1)只读取第一个 block 来推断 schema而不是扫描整个数据集见 dataset.py 的_base_schema实现。如果你明确不想触发任何执行可以传schema(fetch_if_missingFalse)此时未知 schema 返回None——这正是 test_consumption.py 中test_schema用fetch_if_missingFalse断言不产生任何执行任务的原因。此外还有两个高频伴随 APIds.columns()仅返回列名列表如[sepal length (cm), ..., target]同样支持fetch_if_missing参数dataset.pyds.count()返回总行数。对仅由read_parquet创建的 Datasetcount()直接读取 Parquet 元数据计数不读取具体数据非常高效对普通数据集则通过一个轻量的Count逻辑算子统计dataset.py。检查行take() 与 take_all()要拿少量行做内容抽查使用 Dataset.take() 或 Dataset.take_all()。Ray Data 将每一行表示为一个 Python字典import ray ds ray.data.read_csv(s3://anonymousair-example-data/iris.csv) rows ds.take(1) print(rows)输出[{sepal length (cm): 5.1, sepal width (cm): 3.5, petal length (cm): 1.4, petal width (cm): 0.2, target: 0}]两个方法的关键差异与使用要点方法返回默认参数适用场景风险take(limit20)最多limit行组成的 listlimit20快速抽查任意数据集limit过大时数据被拉回调用方机器可能 OOMtake_all(limitNone)全部行组成的 list不设上限只适用于小数据集会把整个数据集拉到调用方机器大数据集必 OOMshow(limit20)无逐行 printlimit20终端里直接看数据同上take_all(limit...)还有一个保护语义如果数据行数超过给定limit会直接抛出ValueError防止你误对大数据集调用。源码中take与take_all都基于iter_rows()逐行产出并组装dataset.py、dataset.py且take内部先执行limit(limit)再迭代因此时间复杂度是 O(limit)。值得注意take()在首次被调用时会打印一条提示日志建议优先用take_batch()以 pandas / numpy 批次格式取数见 dataset.py 的log_once分支。测试 test_take_all 也验证了take_all(4)对 5 行数据集抛出ValueError的行为。拿到行之后可以进一步做行级变换如map、flat_map或逐行迭代参见 Transforming rows 与 Iterating over rows。检查批次take_batch() 与 batch_format行视图适合人工浏览但 Ray Data 的map_batches等算子实际消费的是批次batch——一个 batch 包含多行数据。用 Dataset.take_batch() 可以按下游真正面对的形态来检查数据import ray # 图片数据集默认 batch_format 为 numpy得到 dict[str, np.ndarray] ds ray.data.read_images(s3://anonymousray-example-data/image-datasets/simple) batch ds.take_batch(batch_size2, batch_formatnumpy) print(Batch:, batch) print(Image shape, batch[image].shape)输出示意Batch: {image: array([[[[...]]]], dtypeuint8)} Image shape: (2, 32, 32, 3)batch_format可选值由block.py中的VALID_BATCH_FORMATS定义block.pybatch_format返回类型说明default/numpyDict[str, numpy.ndarray]默认值列名映射到 NumPy 数组pandaspandas.DataFrame适合与 pandas 生态代码衔接pyarrowpyarrow.Table零拷贝、列式存储cudfcudf.DataFrameGPU 加速实验特性一个容易被忽略的关键点batch_format 只决定返回给调用方的表示形式与 Ray Data 底层 block 的存储格式完全无关。也就是说无论内部 block 是 Arrow 还是其他格式你都可以按需指定任一种 batch_format 取数无需关心内部实现。pandas 示例import ray ds ray.data.read_csv(s3://anonymousair-example-data/iris.csv) batch ds.take_batch(batch_size2, batch_formatpandas) print(batch)输出sepal length (cm) sepal width (cm) ... petal width (cm) target 0 5.1 3.5 ... 0.2 0 1 4.9 3.0 ... 0.2 0pyarrow 示例import ray ds ray.data.read_csv(s3://anonymousair-example-data/iris.csv) batch ds.take_batch(batch_size2, batch_formatpyarrow) print(batch)输出pyarrow.Table sepal length (cm): double sepal width (cm): double petal length (cm): double petal width (cm): double target: int64 ---- sepal length (cm): [[5.1,4.9]] sepal width (cm): [[3.5,3]] petal length (cm): [[1.4,1.4]] petal width (cm): [[0.2,0.2]] target: [[0,0]]take_batch() 的源码实现与边界行为take_batch(batch_size20, batch_formatdefault)的实现非常直白dataset.py调用_apply_batch_format()把default解析为DEFAULT_BATCH_FORMAT numpy并校验格式合法性非法值抛出ValueErrorblock.py对数据集先做limit(batch_size)以prefetch_batches0、指定batch_format调用iter_batches()并取第一个批次若数据集为空StopIteration抛出ValueError(The dataset is empty.)。因此它同样有最多返回batch_size行到调用方机器的内存警示batch_size 过大时调用方可能 OOM。测试 test_take_batch 验证了take_batch(3)返回前 3 行、batch_size超过总行数时返回全部行、pandas格式返回pd.DataFrame、numpy 格式返回dict以及空数据集抛ValueError等全部边界。需要注意的是read_images()的 schema图片列默认列名为image类型为ArrowTensorTypeV2(shape(32, 32, 3), dtypeuint8)见 read_api.py所以上面batch[image].shape得到(2, 32, 32, 3)正好是batch 大小 × 高 × 宽 × 通道。拿到批次后若想深入理解批级变换与迭代参见 Transforming batches 与 Iterating over batches。检查执行统计stats() 与日志落盘Ray Data 在执行期间为每个算子统计指标包括墙钟时间wall clock time、CPU 时间、block 变换耗时、峰值堆内存、输出行数/字节数、任务分布与算子吞吐等。对已执行的数据集调用 Dataset.stats() 即可查看import ray from huggingface_hub import HfFileSystem def f(batch): return batch def g(row): return True path hf://datasets/ylecun/mnist/mnist/ fs HfFileSystem() train_files [f[name] for f in fs.ls(path) if train in f[name] and f[name].endswith(.parquet)] ds ( ray.data.read_parquet(train_files, filesystemfs) .map_batches(f) .filter(g) .materialize() ) print(ds.stats())输出示意数值随机器与版本浮动Operator 1 ReadParquet-SplitBlocks(32): 1 tasks executed, 32 blocks produced in 2.92s * Remote wall time: 103.38us min, 1.34s max, 42.14ms mean, 1.35s total * Remote cpu time: 102.0us min, 164.66ms max, 5.37ms mean, 171.72ms total * Block transform time: 95.12us min, 1.31s max, 41.09ms mean, 1.31s total * Peak heap memory usage (MiB): 266375.0 min, 281875.0 max, 274491 mean * Output num rows per block: 1875 min, 1875 max, 1875 mean, 60000 total * Output size bytes per block: 537986 min, 555360 max, 545963 mean, 17470820 total * Output rows per task: 60000 min, 60000 max, 60000 mean, 1 tasks used * Tasks per node: 1 min, 1 max, 1 mean; 1 nodes used * Operator throughput: * Ray Data throughput: 20579.80984833993 rows/s * Estimated single node throughput: 44492.67361278733 rows/s Operator 2 MapBatches(f)-Filter(g): 32 tasks executed, 32 blocks produced in 3.63s * Remote wall time: 675.48ms min, 1.0s max, 797.07ms mean, 25.51s total * Remote cpu time: 673.41ms min, 897.32ms max, 768.09ms mean, 24.58s total * Block transform time: 661.65ms min, 978.04ms max, 778.13ms mean, 24.9s total * Peak heap memory usage (MiB): 152281.25 min, 286796.88 max, 164231 mean * Output num rows per block: 1875 min, 1875 max, 1875 mean, 60000 total * Output size bytes per block: 530251 min, 547625 max, 538228 mean, 17223300 total * Output rows per task: 1875 min, 1875 max, 1875 mean, 32 tasks used * Tasks per node: 32 min, 32 max, 32 mean; 1 nodes used * Operator throughput: * Ray Data throughput: 16512.364546087643 rows/s * Estimated single node throughput: 2352.3683708977856 rows/s Dataset throughput: * Ray Data throughput: 11463.372316361854 rows/s * Estimated single node throughput: 25580.963670075285 rows/s如何解读 stats() 输出对每个物理算子重点看以下几类指标时间指标Remote wall time任务真实墙钟时间、Remote cpu timeCPU 耗时、Block transform timeblock 内数据变换耗时。三者差距大通常意味着存在 IO 等待或调度开销内存指标Peak heap memory usage (MiB)给出 min/max/mean 三个值用于判断是否存在内存尖峰产出指标Output num rows per block、Output size bytes per block、Output rows per task帮助判断 block 划分是否均匀、是否需要调整并行度分布指标Tasks per node反映数据在节点间的分布吞吐指标Ray Data throughput是包含调度开销在内的端到端吞吐Estimated single node throughput是排除分布式开销后的估算值——两者差距大说明分布式调度/网络成为瓶颈可考虑提升任务粒度或数据本地性。两个重要的使用前提stats() 不会触发执行如果数据集尚未执行stats()返回空字符串。必须先通过materialize()、iter_batches()等操作真正执行管线再调用stats()查看源码注释与示例见 dataset.py统计信息会持久化到日志文件每个算子的 stats 同时以日志形式写入/tmp/ray/session_*/logs/ray-data/ray-data.log。该路径由 Ray Data 的日志配置决定——logging.py 中为ray.datalogger 注册了名为ray-data.log的SessionFileHandler落在当前 Ray session 目录下。更完整的观测手段stats()是文本形态的统计快照。若要持续观测Ray Data 还提供进度条、Ray DashboardRay Data Overview 表、Metrics 时间序列视图以及 Prometheus 指标data_output_rows、data_output_bytes、data_cpu_usage_cores、data_gpu_usage_cores等按 dataset/operator 标签区分详见 Monitoring Your Workload。总结与进一步阅读检查数据是 Ray Data 使用流程中成本最低、收益最高的环节ds.schema()O(1)惰性推断与print(ds)回答数据长什么样、有多少行ds.take()/ds.take_all()/ds.show()回答行的内容对不对ds.take_batch(batch_format...)回答下游批次算子的输入形态对不对且返回格式与内部存储解耦ds.stats()回答每个算子花在哪、吞吐与内存如何并落盘到ray-data.log。建议的实践路径读取数据后先schema()take(5)确认结构再在写map_batches前用take_batch(batch_formatpandas)验证批次形态最后对整条管线materialize()后查看stats()定位热点算子。相关主题还可继续阅读 Iterating over data、Transforming data 与 Key concepts。【免费下载链接】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），仅供参考
