Developing a Custom Operator

PreProcess (Data Preprocessing, Optional)

PreProcess is the first step of the data processing workflow, used to preprocess raw data (such as format conversion, deduplication, missing value imputation, and outlier filtering). The execution result serves as the input for Process. PreProcess is defined as needed and must inherit from the Filter, Mapper, or BatchMapper base class. The optional base classes and their applicable scenarios are as follows:

• Inheriting Filter: Used to filter out samples that do not meet the conditions before core processing (such as null values or abnormal formats). The process method returns a bool value to determine whether the sample is retained.
• Inheriting Mapper: Used to perform preprocessing transformations on samples before core processing (such as field renaming or encoding conversion). The process method returns a dict representing newly added or modified columns.
• Inheriting BatchMapper: Used for batch preprocessing scenarios (such as batch feature extraction). The process method returns a pa.Array representing a batch of newly added or modified columns.

Below are code examples for PreProcess:

• Example 1: Inheriting Filter to filter out empty and overly short texts during the preprocessing stage.

  from typing import Any
  
  from modelarts.data.refiner.dataset.dpt import PreTrainText
  from modelarts.data.refiner.op.base_op import Filter
  
  class PreProcess(Filter):
      """Data preprocessing: Filter out empty texts and samples with insufficient length."""
  
      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

• Example 2: Inheriting Mapper to add a text length column to samples during the preprocessing stage.

  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):
      """Data preprocessing: Add a total text length column to samples."""
  
      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 (Core Operator Logic, Mandatory)

Process is the second step of the data processing workflow and a mandatory step, carrying the core processing logic of the operator. Process receives the output of PreProcess as its input (or receives the raw data directly if PreProcess is not defined), executes the core processing, and passes the results to PostProcess. Process must inherit from an operator base class in the SDK (for details on base class differences, see Operator Base Classes and Differences). The optional base classes and their applicable scenarios are as follows:

• Inheriting Mapper: Row-level labeling/transformation. Processes samples one by one and adds new columns. Suitable for scenarios such as text classification labeling, quality scoring, and field transformation. The process method returns dict[str, Any] representing key-value pairs of the newly added columns; it must implement __return_extend_columns_type__() to declare the types of the new columns.
• Inheriting BatchMapper: Batch labeling/transformation. Processes data in batches at the column level. Suitable for scenarios requiring model batch inference, GPU acceleration, etc. The process method receives dict[str, pa.Array] and returns pa.Array; it must implement __return_extend_columns_type__() to declare the types of the new columns.
• Inheriting Filter: Row-level filtering. Evaluates samples one by one to determine whether to retain them. Suitable for scenarios where low-quality samples are filtered out based on conditions. The process method returns a bool (True for retention/False for filtering); it does not need to implement __return_extend_columns_type__().
• Inheriting BatchFilter: Batch filtering. Evaluates samples in batches at the column level to determine whether to retain them. Suitable for batch quality inspection and filtering scenarios. The process method receives dict[str, pa.Array] and returns pa.Array(bool); it does not need to implement __return_extend_columns_type__().
• Inheriting Generator: Generates multiple new samples from one sample. Suitable for scenarios such as video segmentation and data augmentation. The process method returns list[dict[str, Any]], where each element is a complete new sample dictionary; it does not need to implement __return_extend_columns_type__().

Below are code examples for Process:

• Example 1: Inheriting Mapper to assign quality score labels to SFT dialogue data.

  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):
      """Core operator logic: Assign quality score labels to SFT dialogue data."""
  
      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()})

• Example 2: Inheriting Filter to filter out pre-training data with text length below the threshold.

  from typing import Any
  
  from modelarts.data.refiner.dataset.dpt import PreTrainText
  from modelarts.data.refiner.op.base_op import Filter
  
  class Process(Filter):
      """Core operator logic: Filter out pre-training samples whose text length does not exceed the threshold."""
  
      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

• Example 3: Inheriting Generator to split a video into multiple segments.

  from typing import Any
  
  from modelarts.data.refiner.dataset.video import Video
  from modelarts.data.refiner.op.base_op import Generator
  
  class Process(Generator):
      """Core operator logic: Split a video into multiple small clips."""
  
      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]
  
      # Implement video segmentation logic
      def cut_video(self, video_path: str) -> list[Video]:
          ...

PostProcess (Operator Post-Processing, Optional)

PostProcess is the third step of the data processing workflow, used to perform post-processing on the results generated by Process (such as result filtering, format standardization, and redundant column cleanup). PostProcess is defined as needed and must inherit from the Filter, BatchFilter, or Mapper base class. The optional base classes and their applicable scenarios are as follows:

• Inheriting Filter: Used to perform secondary filtering on the results output by Process (such as filtering out low-quality labeling results). The process method returns a bool value to determine whether the sample is retained.
• Inheriting BatchFilter: Used to perform batch secondary filtering on the results output by Process. The process method returns a pa.Array(bool) to determine in batches whether the samples are retained.
• Inheriting Mapper: Used to perform post-processing transformations on the results output by Process (such as field renaming or format standardization). The process method returns a dict representing newly added or modified columns.

Below are code examples for PostProcess:

• Example 1: Inheriting Filter to filter out samples with a quality score of low during the post-processing stage.

  from typing import Any
  
  from modelarts.data.refiner.op.base_op import Filter
  
  class PostProcess(Filter):
      """Operator post-processing: Filter out samples with a 'low' quality score."""
  
      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'

• Example 2: Inheriting BatchFilter to perform batch post-processing and filter out short text samples.

  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):
      """Operator post-processing: Batch filter samples with insufficient text length."""
  
      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_())

Complete process.py Example

Below is a complete process.py example containing all three stages (PreProcess, Process, and PostProcess), demonstrating a "filtering -> labeling -> filtering" pipeline for pre-training text data: PreProcess filters out empty and overly short texts, Process assigns length-level labels to the filtered samples, and PostProcess batch-filters out samples categorized as 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):
    """Step 1 Data preprocessing: Filter out empty texts and overly short texts."""

    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):
    """Step 2 Core operator logic: Assign length-level labels to texts."""

    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):
    """Step 3 Operator post-processing: Batch filter out samples with 'short' level."""

    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_())