# How to Use PipelineBuilder for Declarative ETL Workflows in Semantica

> Learn to build declarative ETL workflows with Semantica's PipelineBuilder. Construct validated, serializable pipelines efficiently using a fluent DSL and explicit dependencies.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: how-to-guide
- Published: 2026-08-21

---

**The `PipelineBuilder` class in `semantica.pipeline` provides a fluent DSL for constructing validated, serializable ETL pipelines by chaining steps with explicit dependencies, configuration, and optional version-control metadata.**

The `semantica-agi/semantica` repository offers a powerful toolkit for defining complex data workflows declaratively. Using the `PipelineBuilder` class located in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py), you can assemble ETL processes as immutable directed acyclic graphs (DAGs) without writing imperative execution logic.

## Core Architecture of the Pipeline Builder

The builder operates on a **declarative** paradigm: you describe *what* the workflow looks like, not *how* the engine executes it. The primary components include:

- **`PipelineBuilder`** – Provides the fluent API to add steps, wire dependencies, configure parallelism, and produce a `Pipeline` dataclass. Implemented in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py).
- **`Pipeline` & `PipelineStep`** – Immutable dataclasses storing step names, types, configurations, and dependency graphs.
- **`PipelineValidator`** – Validates structural integrity, detects circular dependencies, and runs performance heuristics. Located in [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py).
- **`PipelineSerializer`** – Handles serialization to JSON/dict formats for reproducibility and remote execution.

Once built, a `Pipeline` instance can be validated for errors, serialized for storage, or passed to the `ExecutionEngine` for actual processing.

## Defining Steps and Dependencies

The `add_step()` method creates individual pipeline stages with unique identifiers and explicit predecessors.

```python
from semantica.pipeline import PipelineBuilder

builder = PipelineBuilder()
pipeline = (
    builder
    .add_step("ingest", "ingest", source="./documents")
    .add_step(
        "parse",
        "parse",
        formats=["pdf", "docx"],
        dependencies=["ingest"]
    )
    .add_step(
        "embed",
        "embed",
        model="text-embedding-3-large",
        batch_size=32,
        dependencies=["parse"]
    )
    .build(name="DocumentETL")
)

```

Key parameters include:

- **`name`** – Unique step identifier enforced across the pipeline.
- **`type`** – Logical operation category (e.g., `ingest`, `parse`, `embed`).
- **`dependencies`** – List of predecessor step names defining execution order.
- **`handler`** – Optional callable performing the actual work.
- **`delta_mode`, `base_version_id`, `target_version_id`** – Optional version-control flags for tracking data lineage.

For dynamic wiring after step creation, use `connect_steps(from_step, to_step)` to append dependencies programmatically.

## Validation and Serialization

Before execution, validate the pipeline structure to catch configuration errors early:

```python
validation = builder.validate_pipeline()
if not validation["valid"]:
    raise RuntimeError(f"Pipeline validation failed: {validation['errors']}")

```

The `PipelineValidator` checks for duplicate step names, missing dependencies, circular references, and performance anti-patterns. According to the source code in [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py), it returns a dictionary with `valid`, `errors`, and `warnings` keys.

Serialize pipelines for reproducibility or remote deployment:

```python
from semantica.pipeline import PipelineSerializer

serializer = PipelineSerializer()
json_repr = serializer.serialize_pipeline(pipeline, format="json")
restored = serializer.deserialize_pipeline(json_repr)

```

## Practical Implementation Examples

### Basic Fluent API Usage

Construct a complete ETL workflow using method chaining. This example implements a document processing pipeline with normalization and entity extraction:

```python
from semantica.pipeline import PipelineBuilder, ExecutionEngine

builder = PipelineBuilder()
pipeline = (
    builder
    .add_step("ingest", "ingest", source="./documents")
    .add_step(
        "parse",
        "parse",
        formats=["pdf", "docx"],
        dependencies=["ingest"]
    )
    .add_step(
        "normalize",
        "normalize",
        case="lower",
        dependencies=["parse"]
    )
    .add_step(
        "extract",
        "extract",
        entities=True,
        dependencies=["normalize"]
    )
    .add_step(
        "embed",
        "embed",
        model="text-embedding-3-large",
        batch_size=32,
        dependencies=["extract"]
    )
    .build(name="DocumentETL")
)

# Execute

engine = ExecutionEngine()
result = engine.execute_pipeline(pipeline)
print(f"Success: {result.success}")

```

### Building from Declarative Configuration

Load pipeline definitions from YAML or JSON files using `build_pipeline()`:

```python
pipeline_cfg = {
    "name": "YAMLBasedPipeline",
    "steps": [
        {
            "name": "ingest",
            "type": "ingest",
            "config": {"source": "./data"}
        },
        {
            "name": "parse",
            "type": "parse",
            "config": {"formats": ["txt"]},
            "dependencies": ["ingest"]
        },
        {
            "name": "embed",
            "type": "embed",
            "config": {"model": "text-embedding-3-large"},
            "dependencies": ["parse"]
        },
    ],
    "parallelism": 2,
}

builder = PipelineBuilder()
pipeline = builder.build_pipeline(pipeline_cfg)

```

The `build_pipeline()` method (lines 41-90 in [`pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/pipeline_builder.py)) automatically materializes steps from the dictionary structure.

### Using Pipeline Templates

Leverage pre-built templates for common workflows:

```python
from semantica.pipeline import PipelineTemplateManager

tmpl_mgr = PipelineTemplateManager()
builder = tmpl_mgr.create_pipeline_from_template(
    "document_processing",
    ingest={"source": "./docs"},
    parse={"formats": ["pdf", "docx"]},
    embed={"model": "text-embedding-3-large"},
)
pipeline = builder.build()

```

Template definitions reside in [`semantica/pipeline/pipeline_templates.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_templates.py), with usage documentation available in [`semantica/pipeline/pipeline_usage.md`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_usage.md).

## Configuring Parallelism

Set global execution concurrency hints using `set_parallelism()`:

```python
builder = PipelineBuilder()
builder.set_parallelism(level=4)

```

This value instructs the execution engine how many worker threads or processes to utilize when running independent steps.

## Summary

- **`PipelineBuilder`** in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) enables declarative ETL construction through a fluent interface.
- **Steps** are defined via `add_step()` with unique names, types, and optional `dependencies` lists.
- **Validation** through `validate_pipeline()` catches structural errors, circular dependencies, and missing references before runtime.
- **Serialization** via `PipelineSerializer` supports JSON output for reproducible pipeline storage and remote execution.
- **Templates** via `PipelineTemplateManager` provide pre-configured workflows for common patterns like document processing.
- **Configuration-driven** workflows can be instantiated from dictionaries using `build_pipeline()`.

## Frequently Asked Questions

### What is the difference between PipelineBuilder and ExecutionEngine?

**PipelineBuilder** constructs and validates the static pipeline definition (the DAG), while **ExecutionEngine** consumes that definition and orchestrates the actual runtime execution, handling resource scheduling and parallelism.

### How do I detect circular dependencies before runtime?

Call `builder.validate_pipeline()` before execution. The `PipelineValidator` class in [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py) performs topological sorting and detects cycles, returning specific error messages identifying the problematic dependency chain.

### Can I serialize a pipeline to JSON and restore it later?

Yes. Use `PipelineSerializer.serialize_pipeline(pipeline, format="json")` to generate a JSON string, then call `deserialize_pipeline()` to reconstruct the `Pipeline` dataclass. This enables version-controlled workflow definitions and distributed execution.

### How does version control (delta mode) work in pipeline steps?

When adding steps, you can specify `delta_mode=True` along with `base_version_id` and `target_version_id` parameters. These flags are stored in the step configuration and consumed by downstream components in [`semantica/pipeline/pipeline_provenance.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_provenance.py) to track data lineage and incremental processing states.