更新时间：2026-09-28 GMT+08:00

算子基类与差异详解

在开发自定义算子时，您需要选择合适的算子基类来继承。不同的基类适用于不同的处理场景，如行级转换、批量过滤、一对多生成等。本文详细介绍SDK提供的5种算子基类的继承体系、关键参数、差异对比及选型建议，帮助您快速选择合适的基类。

算子基类说明

继承是面向对象编程的核心概念：子类自动获得父类的所有属性和方法，同时可以重写（覆盖）父类的方法来实现自定义逻辑。自定义算子通过继承SDK基类，只需实现process方法即可完成数据处理逻辑的开发。

OP (父基类，不可直接使用)
├── Mapper         — 行级转换/打标算子
├── BatchMapper    — 批量转换/打标算子
├── Filter         — 行级过滤算子
├── BatchFilter    — 批量过滤算子
└── Generator      — 一生多算子（一个样本生成多个样本）

各算子基类的关键信息对比如下表所示。OP父基类的参数说明请参见OP父基类参数说明。

算子基类	处理粒度	典型场景	PreProcess （可选）	Process （必填）	PostProcess （可选）	process方法输入	process方法返回值	是否需声明新列类型（可选）
Filter	单条样本	行级过滤	支持	支持	支持	dict[str, Any]	bool	否
Mapper	单条样本	行级打标/转换	支持	支持	支持	dict[str, Any]	dict[str, Any]（可新增一列或多列）	是（重写__return_extend_columns_type__()）
BatchMapper	批量样本	批量打标/转换	支持	支持	不支持	dict[str, pa.Array]	pa.Array（多列时使用pa.struct类型）	是（重写__return_extend_columns_type__()）
BatchFilter	批量样本	批量过滤	不支持	支持	支持	dict[str, pa.Array]	pa.Array(bool)	否
Generator	单条样本	一生多	不支持	支持	不支持	dict[str, Any]	list[dict[str, Any]]	否

OP父基类参数说明

OP是所有算子的顶层基类，不可直接继承使用。定义了算子的公共属性和方法，关键参数说明如下表所示。

参数名	类型	来源	说明
op_args	dict[str, Any]	用户自定义	用户自定义的算子参数。创建算子时通过界面配置传入，代码中通过__init__获取。
content_type	str	系统注入	当前数据集类型，对应DatasetType枚举值。
input_columns	list[str]	系统注入	算子处理的输入列名列表，无需在op_args中定义。
_accelerator	str	开发者定义	加速器类型，可选值："cpu"（默认）或"npu"。
_max_concurrency	int	开发者定义	最大并发数。调小避免显存溢出，调大提升吞吐。

选型建议

选择基类的决策流程：
  我要转换/打标数据 → Mapper（行级）或 BatchMapper（批量）
  我要过滤数据 → Filter（行级）或 BatchFilter（批量）
  我要一条生成多条 → Generator
行级 vs 批量：
  无需模型推理 → 行级（Mapper/Filter），逻辑简单
  需要模型推理 → 批量（BatchMapper/BatchFilter），吞吐更高

行级与批量算子的选择建议：

行级算子（Mapper/Filter）：逐条处理，逻辑简单，适合无需批量推理的场景。
批量算子（BatchMapper/BatchFilter）：以列为单位批量处理，适合需要模型批量推理场景，可显著提升吞吐。

数据集类型与字段

SDK内置了以下数据集实体类，用于在算子中解析和构造样本数据：

数据集类型	字段	引入方式
PreTrainText（预训练文本）	text（str）	from modelarts.data.refiner.dataset.dpt import PreTrainText
Messages（OpenAI Messages格式）	messages（List[Message]），每个Message包含role（str）和content（str）	from modelarts.data.refiner.dataset.sft import Messages, Message
Image（图片）	file_path（str）、record（Optional[bytes]）、height（int）、width（int）	from modelarts.data.refiner.dataset.image import Image
Video（视频）	file_path（str）、duration（float）、fps（float）、height（int）、width（int）、frame_count（int）	from modelarts.data.refiner.dataset.video import Video

每种数据集实体类均提供from_dict(sample)将字典转为对象、to_dict(obj)将对象转为字典、check(sample)校验样本格式三个静态方法。

常见问题

process.py中可以定义多个类吗？
可以。但只有PreProcess、Process、PostProcess这三个类名会被识别和执行。您可以定义辅助类和函数供这三个类调用。
PreProcess和PostProcess可以继承不同的基类吗？
可以。三个处理阶段可以继承不同的算子基类，例如PreProcess继承Filter（过滤），Process继承Mapper（打标）。
Mapper和BatchMapper的process返回值有什么区别？
Mapper的process返回dict[str, Any]，表示一条样本新增列的键值对；BatchMapper的process返回pa.Array，表示一整列批量数据的结果。
__return_extend_columns_type__()什么时候需要实现？
仅当算子类型为Mapper或BatchMapper时需要实现，用于声明新增列的类型。Filter、BatchFilter、Generator不需要实现此方法。

如何声明多列返回类型？

@staticmethod 
def __return_extend_columns_type__() -> daft.schema.Schema: 
    return daft.schema.Schema.from_pydict({ 
        'score': daft.DataType.int64(), 
        'label': daft.DataType.string(),
        'confidence': daft.DataType.float64(),
    })

Generator的返回值和其他算子有什么区别？
Generator的process返回list[dict[str, Any]]，每个元素是一条完整的新样本（包含所有列），而Mapper只需返回新增列。Generator适用于一条样本生成多条的场景。

父主题： 开发自定义算子