Extractor

The extractor is the first stage of the pipeline. It reads source files and emits one envelope per file containing extracted data. A pipeline has exactly one extractor, and the extractor determines which file format the pipeline ingests.

Usage

Configured under the top-level extractor key.

extractor:
  name: i3
  kwargs:
    gcd_path: /path/to/GCD.i3.zst
    include: [ ... ]
    ml_suite: { ... }

How it Works

The extractor handles data extraction from disk to memory. For each file the extractor reads the relevant frames and emits an envelope of raw data downstream, where the collector then packs files into shards.

Each envelope must carry:

• the file it was read from, under attrs.LOCAL.origin. This is handled by default.

• the ids of every event it holds, as Envelope.events: one row per event, its columns being the ones that identify an event. Processors key on these columns by default.

A file without events should still be emitted, with an empty events frame, so it is counted in a shard. Return None only for a file that must not count, which is then left out as if it were never given.

Variants

• I3: extracts features from IceCube I3 files.

Registering a new extractor

An extractor is a subclass of Extractor that declares a name, a version, and the file_ext it reads, and implements the per-file extraction. Register it with ExtractorFactory.

from typing import Any, ClassVar

from icegraph.data.extractor import Extractor, ExtractorFactory
from icegraph.data.envelope import Envelope

from .config import MyExtractorConfig

class MyExtractor(Extractor[MyExtractorConfig]):
    name: ClassVar[str] = "my-extractor"
    version: ClassVar[int] = 1
    file_ext: ClassVar[str] = ".myext"

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

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

    def _process(self, item) -> Envelope | None:
        ...  # read the file and emit an envelope

ExtractorFactory.register(MyExtractor)