Processor

A processor is a single transformation stage in the pipeline. The pipeline runs an ordered list of processors between the extractor and the writer, each applying one operation to the data as it streams through. Processors are where feature engineering happens: selecting and renaming columns, building graph edges, computing weights, assigning dataset splits, etc.

Usage

The processors key holds an ordered list; each entry selects a processor by name. Order matters, since each processor sees the result of the previous one.

processors:
  - name: select
    kwargs: { key: pulses }
  - name: knn
    kwargs: { by: [string, om], col: position, out: [edge_index, edge_attr], k: 8 }

How it Works

The data flowing between stages is an envelope holding the raw extracted frames, a set of working frames, and the committed output. Processors operate on the active working frame: the select processor chooses which frame is active, most processors transform that frame in place, and the commit processor writes finished columns into the output that the writer persists. Columns can be addressed individually or through named groups defined by the alias processor.

Variants

Staging:

• select: choose the active working frame.

• commit: write finished columns to the output.

• copy: copy columns into another frame.

Columns and values:

• alias: define named groups of columns.

• rename: rename columns.

• map: remap the values of a column.

• fill: add or overwrite a column with a constant.

• unique: record the distinct values of columns.

• stats: compute per-column statistics.

Graph construction:

• domproc: convert DOM identifiers to positions.

• knn: build k-nearest-neighbor graph edges.

• compress: stack rows into per-event arrays.

• pivot: reshape long-form data to wide.

Weighting and splitting:

• simweights: compute event weights.

• splitmap: assign train/validation/test splits.

Debugging:

• inspect: print the active frame for inspection.

Registering a new processor

A processor is a subclass of Processor that declares a name and version and implements the transformation:

_process(self, item) -> Envelope | None

Transform the envelope and return it, or return None to drop the event.

from typing import Any, ClassVar

from icegraph.data.processor import Processor, ProcessorFactory
from icegraph.data.envelope import Envelope

from .config import MyProcessorConfig

class MyProcessor(Processor[MyProcessorConfig]):
    name: ClassVar[str] = "my-processor"
    version: ClassVar[int] = 1

    @classmethod
    def validate_config(cls, config: dict[str, Any]) -> MyProcessorConfig:
        return MyProcessorConfig(**config)

    def build(self) -> None:
        ...

    def _process(self, item: Envelope) -> Envelope | None:
        ...

ProcessorFactory.register(MyProcessor)