vcti-data-pipeline 1.1.0

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

Data Pipeline

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

Installation

pip install vcti-data-pipeline>=1.0.0

In pyproject.toml dependencies

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

---

Quick Start

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

# 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 DataRecord(data=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.data.copy()
        scaled["value"] *= self.factor
        return DataRecord(data=scaled, attributes=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.data 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

Type	Inputs	Purpose
DataSource	0 (leaf)	Load data from external source
DataTransformer	1	Transform one record to another
DataReducer	1+	Combine multiple records into one
DataCombiner	1+	Merge arrays field-wise (extends DataReducer)
DataTarget	1	Persist record, pass through
ForEachFlow	—	Iterate over keyed cases

Other types

Type	Purpose
DataRecord	Alias for DataNode — the data payload flowing between nodes
Column	Frozen dataclass coupling column name with numpy dtype
FlowNode	Abstract base for all flow nodes

---

Dependencies

• numpy (>=1.24)
• vcti-array-tree (>=1.0.0) — DataNode (aliased as DataRecord)