# How to Build Declarative ETL Pipelines with the Semantica Pipeline DSL

> Build declarative ETL pipelines effortlessly using Semantica's pipeline DSL. Automate dependency resolution, parallel execution, and retries for robust data workflows.

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

---

**Semantica's pipeline DSL lets you define end-to-end ETL workflows as composable, serializable steps that handle dependency resolution, parallel execution, retries, and progress tracking automatically.**

In the `semantica-agi/semantica` open-source repository, the pipeline DSL transforms imperative Python scripts into maintainable, reproducible data pipelines. The system separates *what* your pipeline does (declarative step definitions) from *how* it runs (parallelism, failure handling, scheduling), enabling you to build robust ETL workflows without boilerplate orchestration code.

---

## Core Components of the Pipeline DSL

The Semantica pipeline system consists of layered components, each with a distinct responsibility:

| Component | Purpose | Source File |
|-----------|---------|-------------|
| `PipelineBuilder` | Fluent DSL for constructing pipelines with `add_step`, `connect_steps`, and `build` | [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) |
| `PipelineStep` | Dataclass representing a single step: name, type, handler, dependencies, status, and delta mode support | `semantica/pipeline/pipeline_builder.py#L55` |
| `PipelineValidator` | Pre-execution validation for cycles, missing handlers, and misconfigured connections | [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py) |
| `ExecutionEngine` | Runtime orchestration: scheduling, resource allocation, progress tracking, and result aggregation | [`semantica/pipeline/execution_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/execution_engine.py) |
| `FailureHandler` | Per-step retry policies (exponential, linear, fixed) and failure strategies (skip, retry, abort, fallback) | [`semantica/pipeline/failure_handler.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/failure_handler.py) |
| `ParallelismManager` | Thread or process pool management for concurrent step execution | [`semantica/pipeline/parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/parallelism_manager.py) |
| `PipelineSerializer` | YAML serialization for pipeline reproducibility across environments | `semantica/pipeline/pipeline_builder.py#L88` |
| `ExecutionResult` | Post-execution metadata: success status, output, timing metrics, and error details | `semantica/pipeline/execution_engine.py#L62` |

---

## Step 1: Constructing a Pipeline with PipelineBuilder

The `PipelineBuilder` class in [`semantica/pipeline/pipeline_builder.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_builder.py) provides a **fluent interface** for defining steps and their dependencies. Each `add_step` call creates a `PipelineStep` dataclass; `connect_steps` establishes the directed acyclic graph (DAG) that determines execution order.

```python
from semantica.pipeline import PipelineBuilder
from semantica.ingest import FileIngestor
from semantica.parse import DocumentParser
from semantica.semantic_extract import NERExtractor
from semantica.kg import GraphBuilder

# Initialize handlers

ingestor   = FileIngestor()
parser     = DocumentParser()
extractor  = NERExtractor(method="ml")
kg_builder = GraphBuilder(merge_entities=True)

# Build the pipeline declaratively

builder = PipelineBuilder()
builder.add_step("ingest",   "file_ingest",    handler=ingestor.ingest_file)
builder.add_step("parse",    "document_parse", handler=parser.parse)
builder.add_step("extract",  "ner_extract",    handler=extractor.extract)
builder.add_step("build_kg", "graph_build",    handler=kg_builder.build)

# Define data flow: ingest → parse → extract → build_kg

builder.connect_steps("ingest", "parse")
builder.connect_steps("parse",  "extract")
builder.connect_steps("extract", "build_kg")

pipeline = builder.build("my_etl_pipeline")

```

The `handler` parameter accepts any **Python callable**, making the DSL agnostic to your implementation. Step names passed to `add_step` are internal identifiers; the second argument is the step type used for logging and metrics.

---

## Step 2: Validating Pipeline Structure

Before execution, validate your pipeline with `PipelineValidator` to catch **cycles, orphaned steps, and missing handlers** early. The validator in [`semantica/pipeline/pipeline_validator.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/pipeline_validator.py) returns a structured result with `valid`, `errors`, and `warnings` fields.

```python
from semantica.pipeline import PipelineValidator

validator = PipelineValidator()
validation = validator.validate_pipeline(pipeline)

if not validation.valid:
    for err in validation.errors:
        print(f"Validation error: {err}")
    raise RuntimeError("Pipeline validation failed")

for warn in validation.warnings:
    print(f"Warning: {warn}")

```

This validation step prevents runtime failures from structural issues like circular dependencies between steps.

---

## Step 3: Configuring Parallelism and Failure Handling

The `ExecutionEngine` in [`semantica/pipeline/execution_engine.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/execution_engine.py) orchestrates runtime behavior. Control **concurrency** via `max_workers` and **resilience** via `FailureHandler` policies.

```python
from semantica.pipeline import (
    ExecutionEngine,
    RetryPolicy,
    RetryStrategy,
    FailureHandler
)

# Initialize engine with 4 parallel workers

engine = ExecutionEngine(max_workers=4)

# Configure exponential backoff for the NER extraction step

policy = RetryPolicy(
    max_retries=3,
    strategy=RetryStrategy.EXPONENTIAL,
    initial_delay=1.0,
    backoff_factor=2.0,  # 1s, 2s, 4s delays

)

handler = FailureHandler()
handler.set_retry_policy("ner_extract", policy)

# Inject custom failure handling

engine.failure_handler = handler

# Execute with input data path

result = engine.execute_pipeline(pipeline, data="data/")

print(f"Success: {result.success}")
print(f"Execution time: {result.metrics['execution_time']:.1f}s")
print(f"Steps executed: {result.metrics['steps_executed']}")
print(f"Steps failed: {result.metrics['steps_failed']}")

```

The `ParallelismManager` ([`semantica/pipeline/parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/parallelism_manager.py)) automatically schedules steps with no pending dependencies onto the worker pool. Steps become eligible for execution as soon as all their upstream dependencies complete successfully.

---

## Step 4: Enabling Reproducibility with YAML Serialization

Pipelines built with the DSL are **fully serializable**. The `PipelineSerializer` enables version-controlled, environment-agnostic pipeline definitions.

```python

# Save to YAML

yaml_path = "my_pipeline.yaml"
builder.serialize(pipeline, output_path=yaml_path)

# Load and replay on any machine

from semantica.pipeline import PipelineSerializer

loaded_pipeline = PipelineSerializer().load(yaml_path)

```

This capability is critical for **CI/CD integration**, where the same pipeline definition must run identically across development, staging, and production environments.

---

## Step 5: Implementing Incremental Processing with Delta Mode

For large datasets, enable **delta mode** to process only changed records between pipeline runs. Configure `delta_mode=True` with version identifiers:

```python
builder.add_step(
    "extract",
    "ner_extract",
    handler=extractor.extract,
    delta_mode=True,
    base_version_id="v1.2",
    target_version_id="v1.3",
)

```

The `ExecutionEngine` feeds only documents whose version changed between `base_version_id` and `target_version_id` to the handler, significantly reducing reprocessing time for incremental ETL workflows.

---

## Execution Flow Summary

Understanding the complete lifecycle helps debug and optimize pipelines:

1. **Construction** — `PipelineBuilder.add_step()` and `connect_steps()` define the DAG
2. **Validation** — `PipelineValidator.validate_pipeline()` catches structural errors
3. **Serialization (optional)** — `builder.serialize()` produces reproducible YAML
4. **Execution** — `ExecutionEngine.execute_pipeline()` schedules steps, manages the `ParallelismManager` pool, and applies `FailureHandler` policies
5. **Result** — `ExecutionResult` provides success status, output data, timing metrics, and error details

---

## Summary

- Use `PipelineBuilder` for **declarative step definition** with a fluent, chainable API
- Validate pipelines with `PipelineValidator` before execution to catch cycles and missing handlers
- Control **parallelism** via `ExecutionEngine(max_workers=n)` and the `ParallelismManager`
- Implement **fault tolerance** through `FailureHandler` retry policies (exponential, linear, fixed) and strategies (skip, retry, abort, fallback)
- Ensure **reproducibility** by serializing pipelines to YAML with `PipelineSerializer`
- Reduce reprocessing overhead with **delta mode** for incremental ETL runs

---

## Frequently Asked Questions

### What makes Semantica's pipeline DSL "declarative"?

The DSL separates *what* from *how*. You declare steps and their relationships; the `ExecutionEngine` handles scheduling, parallelism, retries, and resource management. This contrasts with imperative frameworks where you manually coordinate execution flow.

### How does the DSL handle step failures?

The `FailureHandler` class in [`semantica/pipeline/failure_handler.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/failure_handler.py) supports per-step policies. You can configure retries with exponential backoff, skip failed steps, abort the entire pipeline, or execute fallback handlers. Unhandled exceptions propagate to `ExecutionResult.errors`.

### Can I mix thread and process parallelism?

The `ParallelismManager` in [`semantica/pipeline/parallelism_manager.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/pipeline/parallelism_manager.py) abstracts pool creation. Configure the pool type through `ExecutionEngine` parameters; the DSL itself remains agnostic to the underlying concurrency model. CPU-bound steps benefit from process pools; I/O-bound steps from thread pools.

### Is the pipeline format portable across Python versions?

YAML-serialized pipelines are **version-agnostic configurations**. The serialized format contains step metadata, dependencies, and handler references—not serialized code. Handlers must exist in the target environment, but the pipeline structure transfers without compatibility issues.