Data Pipeline

ETL-style data processing framework with composable flow nodes — sources, transformers, reducers, and combiners.

Installation

pip install vcti-data-pipeline>=2.0.0

In pyproject.toml dependencies

dependencies = [
    "vcti-data-pipeline>=2.0.0",
]

Quick Start

import numpy as np
from vcti.pipeline import DataSource, DataTransformer, DataRecord, data_record

# 1. Define a source
class CsvSource(DataSource):
    def __init__(self, path: str, *, name: str | None = None) -> None:
        super().__init__(name=name)
        self.path = path

    def load(self) -> DataRecord:
        arr = np.genfromtxt(self.path, dtype=None, delimiter=",", names=True)
        return data_record(arr)

# 2. Define a transformer
class ScaleTransformer(DataTransformer):
    def __init__(self, factor: float, *, name: str | None = None) -> None:
        super().__init__(name=name)
        self.factor = factor

    def transform(self, record: DataRecord) -> DataRecord:
        scaled = record.load().copy()
        scaled["value"] *= self.factor
        return data_record(scaled, record.attributes)

# 3. Compose the flow (right-to-left)
source = CsvSource("data.csv", name="raw-csv")
scaler = ScaleTransformer(factor=2.0, name="2x-scaler")
flow = scaler.connect(source)

# 4. Execute
result = flow.execute()

Combining multiple sources

from vcti.pipeline import DataCombiner

ids = IdSource()
coords = CoordSource()

combiner = DataCombiner()
flow = combiner.connect(ids)
flow = flow.connect(coords)

combined = flow.execute()
# combined.load() has fields from both sources

Iterating over cases

from vcti.pipeline import ForEachFlow

flow = ForEachFlow(
    keys=CasesSource(reader),
    factory=lambda case_id: CaseDataSource(reader, case_id),
    key_field="ID",
)

for case_id, case_flow in flow:
    record = case_flow.execute()
    # process each case...

Flow Node Types

Other types

Dependencies

• numpy (>=1.24)
• vcti-datanode (>=2.0.0) — DataNode (aliased as DataRecord), plus EagerDataSource / LazyDataSource