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.

The envelope a processor receives is a collected block of one or more input files, processed as if it were one file.

Event ids

The extractor sets the columns identifying an event on the envelope. Every option naming them defaults to the ID columns defined here. __ids__ can be used to reference them anywhere columns are requested:

- name: pivot
  kwargs:
    index: [ __ids__, string, om, pmt ]
    col: vector_index
    values: item

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.

  • i3-simulation: store the simulation a file was generated with and the per-event quantities to weight it.

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.

  • 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)