开发自定义算子

PreProcess（数据预处理，可选）

PreProcess是数据处理流程的第1步，用于对原始数据进行预处理（如格式转换、去重、缺失值填充、异常值过滤等），执行结果作为Process的输入。PreProcess为按需定义，必须继承Filter、Mapper或BatchMapper基类。可选基类及适用场景如下：

• 继承Filter：用于在核心处理之前过滤掉不符合条件的样本（如空值、格式异常等），process方法返回bool决定样本是否保留。
• 继承Mapper：用于在核心处理之前对样本进行预处理转换（如字段重命名、编码转换等），process方法返回dict为新增或修改的列。
• 继承BatchMapper：用于批量预处理场景（如批量特征提取），process方法返回pa.Array为批量新增或修改的列。

PreProcess代码示例如下：

• 示例1：继承Filter，预处理阶段过滤掉空文本和过短文本。

  from typing import Any
  
  from modelarts.data.refiner.dataset.dpt import PreTrainText
  from modelarts.data.refiner.op.base_op import Filter
  
  class PreProcess(Filter):
      """数据预处理：过滤掉空文本和长度不足的样本"""
  
      def __init__(self, op_args: dict[str, Any], **kwargs):
          super().__init__(op_args, **kwargs)
          self.min_length = op_args.get("min_length", 50)
  
      def process(self, sample: dict[str, Any]) -> bool:
          pre_train_text = PreTrainText.from_dict(sample)
          if not pre_train_text.text or not pre_train_text.text.strip():
              return False
          return len(pre_train_text.text) >= self.min_length

• 示例2：继承Mapper，预处理阶段为样本新增文本长度列。

  from typing import Any
  
  import daft
  from modelarts.data.refiner.dataset.sft import Messages
  from modelarts.data.refiner.op.base_op import Mapper
  
  class PreProcess(Mapper):
      """数据预处理：为样本新增文本总长度列"""
  
      def __init__(self, op_args: dict[str, Any], **kwargs):
          super().__init__(op_args, **kwargs)
  
      def process(self, sample: dict[str, Any]) -> dict[str, Any]:
          messages = Messages.from_dict(sample)
          total_len = sum(len(msg.content) for msg in messages.messages)
          return {'text_length': total_len}
  
      @staticmethod
      def __return_extend_columns_type__() -> daft.schema.Schema:        return daft.schema.Schema.from_pydict({'text_length': daft.DataType.int64()})

Process（算子核心逻辑，必填）

Process是数据处理流程的第2步，也是必填步骤，承载算子的核心处理逻辑。Process接收PreProcess的输出作为输入（如果未定义PreProcess，则接收原始数据），执行核心处理后将结果传递给PostProcess。Process必须继承SDK中的算子基类（关于基类差异详见算子基类与差异详解），可选基类及适用场景如下：

• 继承Mapper：行级打标/转换，逐条处理样本并新增列。适用于文本分类打标、质量评分、字段变换等场景。process方法返回dict[str, Any]为新增列的键值对；必须实现__return_extend_columns_type__()声明新增列类型。
• 继承BatchMapper：批量打标/转换，以列为单位批量处理。适用于需要模型批量推理、GPU加速等场景。process方法接收dict[str, pa.Array]，返回pa.Array；必须实现__return_extend_columns_type__()声明新增列类型。
• 继承Filter：行级过滤，逐条判断样本是否保留。适用于按条件过滤低质量样本的场景。process方法返回bool，True保留/False过滤；无需实现__return_extend_columns_type__()。
• 继承BatchFilter：批量过滤，以列为单位批量判断样本是否保留。适用于批量质量检测过滤的场景。process方法接收dict[str, pa.Array]，返回pa.Array(bool)；无需实现__return_extend_columns_type__()。
• 继承Generator：一条样本生成多条新样本。适用于视频切分、数据增强等场景。process方法返回list[dict[str, Any]]，每个元素是完整的新样本字典；无需实现__return_extend_columns_type__()。

Process代码示例如下：

• 示例1：继承Mapper，为SFT对话数据打质量评分标签。

  from typing import Any
  
  import daft
  from modelarts.data.refiner.dataset.sft import Messages
  from modelarts.data.refiner.op.base_op import Mapper
  
  class Process(Mapper):
      """算子核心逻辑：为SFT对话数据打质量评分标签"""
  
      def __init__(self, op_args: dict[str, Any], **kwargs):
          super().__init__(op_args, **kwargs)
          self.high_threshold = op_args.get("high_threshold", 5000)
          self.medium_threshold = op_args.get("medium_threshold", 1000)
  
      def process(self, sample: dict[str, Any]) -> dict[str, Any]:
          messages = Messages.from_dict(sample)
          total_len = sum(len(msg.content) for msg in messages.messages)
          if total_len > self.high_threshold:
              return {'quality': 'high'}
          elif total_len > self.medium_threshold:
              return {'quality': 'medium'}
          else:
              return {'quality': 'low'}
  
      @staticmethod
      def __return_extend_columns_type__() -> daft.schema.Schema:
          return daft.schema.Schema.from_pydict({'quality': daft.DataType.string()})

• 示例2：继承Filter，过滤文本长度不足的预训练数据。

  from typing import Any
  
  from modelarts.data.refiner.dataset.dpt import PreTrainText
  from modelarts.data.refiner.op.base_op import Filter
  
  class Process(Filter):
      """算子核心逻辑：过滤文本长度不超过阈值的预训练样本"""
  
      def __init__(self, op_args: dict[str, Any], **kwargs):
          super().__init__(op_args, **kwargs)
          self.text_len = op_args.get("text_len", 1000)
  
      def process(self, sample: dict[str, Any]) -> bool:
          pre_train_text = PreTrainText.from_dict(sample)
          return len(pre_train_text.text) > self.text_len

• 示例3：继承Generator，视频切分生成多个片段。

  from typing import Any
  
  from modelarts.data.refiner.dataset.video import Video
  from modelarts.data.refiner.op.base_op import Generator
  
  class Process(Generator):
      """算子核心逻辑：将视频切分为多个小片段"""
  
      def __init__(self, op_args: dict[str, Any], **kwargs):
          super().__init__(op_args, **kwargs)
          self.clip_duration = op_args.get("clip_duration", 10)
  
      def process(self, sample: dict[str, Any]) -> list[dict[str, Any]]:
          video = Video.from_dict(sample)
          video_clips = self.cut_video(video.file_path)
          return [Video.to_dict(clip) for clip in video_clips]
  
      # 实现视频切分逻辑
      def cut_video(self, video_path: str) -> list[Video]:
          ...

PostProcess（算子后处理，可选）

PostProcess是数据处理流程的第3步，用于对Process处理后的结果进行后处理（如结果过滤、格式标准化、冗余列清理等）。PostProcess为按需定义，必须继承Filter、BatchFilter或Mapper基类。可选基类及适用场景如下：

• 继承Filter：用于对Process输出的结果进行二次过滤（如过滤低质量打标结果），process方法返回bool决定样本是否保留。
• 继承BatchFilter：用于对Process输出的结果进行批量二次过滤，process方法返回pa.Array(bool)批量决定样本是否保留。
• 继承Mapper：用于对Process输出的结果进行后处理转换（如字段重命名、格式标准化等），process方法返回dict为新增或修改的列。

PostProcess代码示例如下：

• 示例1：继承Filter，后处理阶段过滤掉质量评分为low的样本。

  from typing import Any
  
  from modelarts.data.refiner.op.base_op import Filter
  
  class PostProcess(Filter):
      """算子后处理：过滤掉质量评分为low的样本"""
  
      def __init__(self, op_args: dict[str, Any], **kwargs):
          super().__init__(op_args, **kwargs)
  
      def process(self, sample: dict[str, Any]) -> bool:
          quality = sample.get('quality', 'low')
          return quality != 'low'

• 示例2：继承BatchFilter，批量后处理过滤短文本样本。

  from typing import Any
  
  import pyarrow as pa
  from modelarts.data.refiner.dataset.dpt import TEXT
  from modelarts.data.refiner.op.base_op import BatchFilter
  
  class PostProcess(BatchFilter):
      """算子后处理：批量过滤文本长度不足的样本"""
  
      def __init__(self, op_args: dict[str, Any], **kwargs):
          super().__init__(op_args, **kwargs)
          self.text_len = op_args.get("text_len", 100)
  
      def process(self, sample: dict[str, pa.Array]) -> pa.Array:
          texts = sample.get(TEXT)
          results = []
          for text in texts:
              text_value = text.as_py()
              results.append(len(text_value) > self.text_len)
              return pa.array(results, type=pa.bool_())

完整process.py示例

以下是一个包含PreProcess、Process、PostProcess三个阶段的完整process.py示例，演示对预训练文本数据的“过滤→打标→过滤”流水线：PreProcess过滤空文本和过短文本，Process为通过过滤的样本打上长度等级标签，PostProcess批量过滤掉short级别的样本。

from typing import Any

import daft
import pyarrow as pa
from modelarts.data.refiner.dataset.dpt import PreTrainText
from modelarts.data.refiner.op.base_op import Filter, Mapper, BatchFilter

class PreProcess(Filter):
    """第1步数据预处理：过滤掉空文本和过短文本"""

    def __init__(self, op_args: dict[str, Any], **kwargs):
        super().__init__(op_args, **kwargs)
        self.min_length = op_args.get("min_length", 50)

    def process(self, sample: dict[str, Any]) -> bool:
        pre_train_text = PreTrainText.from_dict(sample)
        if not pre_train_text.text or not pre_train_text.text.strip():
            return False
        return len(pre_train_text.text) >= self.min_length

class Process(Mapper):
    """第2步算子核心逻辑：为文本打上长度等级标签"""

    def __init__(self, op_args: dict[str, Any], **kwargs):
        super().__init__(op_args, **kwargs)

    def process(self, sample: dict[str, Any]) -> dict[str, Any]:
        pre_train_text = PreTrainText.from_dict(sample)
        text_len = len(pre_train_text.text)
        if text_len > 5000:
            level = "long"
        elif text_len > 500:
            level = "medium"
        else:
            level = "short"
        return {"length_level": level}

    @staticmethod
    def __return_extend_columns_type__() -> daft.schema.Schema:
        return daft.schema.Schema.from_pydict({"length_level": daft.DataType.string()})

class PostProcess(BatchFilter):
    """第3步算子后处理：批量过滤掉short级别的样本"""

    def __init__(self, op_args: dict[str, Any], **kwargs):
        super().__init__(op_args, **kwargs)

    def process(self, sample: dict[str, pa.Array]) -> pa.Array:
        levels = sample.get("length_level")
        results = []
        for level in levels:
            results.append(level.as_py() != "short")
            return pa.array(results, type=pa.bool_())