How to Build Declarative ETL Pipelines with the Semantica Pipeline DSL

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
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
ExecutionEngine	Runtime orchestration: scheduling, resource allocation, progress tracking, and result aggregation	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
ParallelismManager	Thread or process pool management for concurrent step execution	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 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.

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 returns a structured result with valid, errors, and warnings fields.

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 orchestrates runtime behavior. Control concurrency via max_workers and resilience via FailureHandler policies.

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) 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.

# 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:

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 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}