Pipeline

glossary

Overview

A pipeline is one of the most pervasive abstractions in software engineering: a sequence of processing stages connected end-to-end, where the output of one stage becomes the input of the next. Pipelines appear everywhere -- from Unix shell commands piped together with |, to CI/CD build chains, to real-time data streaming architectures processing millions of events per second. The pipeline metaphor draws directly from manufacturing assembly lines, where raw materials flow through specialized stations, each adding value, until a finished product emerges at the end.

In the Prismatic Platform, pipelines are not merely a convenience pattern but a foundational architectural principle. The platform employs pipelines at every level: 16-level epistemic reasoning pipelines that process beliefs through quality gates, SEADF evolution pipelines that drive autonomous improvement, 11-phase pre-commit pipelines that enforce code quality, and Broadway-based concurrent data pipelines that process OSINT intelligence feeds. Understanding pipelines -- their patterns, failure modes, and composition strategies -- is essential to working effectively with the platform.

Definition and the Pipeline Metaphor

A pipeline, in its most general form, is a chain of processing elements arranged so that the output of each element feeds directly into the input of the next. The term originates from plumbing: physical pipes that carry fluid from one location to another. In computing, the metaphor was first applied to CPU instruction pipelines in the 1960s and later popularized by Doug McIlroy's Unix pipes in the 1970s.

Three properties distinguish a pipeline from a general workflow:

  1. Sequential ordering -- stages execute in a defined order, each depending on its predecessor's output.
  2. Data transformation -- each stage receives data, transforms it, and passes the result forward.
  3. Composability -- stages can be added, removed, or reordered independently of each other.

The assembly line metaphor is instructive: Henry Ford's innovation was not any single manufacturing step but the arrangement of steps into a flow. Similarly, pipeline power comes not from individual stages but from their composition. A Unix command like cat access.log | grep 404 | cut -d' ' -f1 | sort | uniq -c | sort -rn composes six simple tools into a powerful log analysis pipeline that none could accomplish alone.

Historical Context

The pipeline concept has deep roots across multiple computing domains:

  • 1960s -- CPU instruction pipelines: IBM System/360 introduced instruction pipelining, overlapping fetch, decode, and execute stages to increase throughput. This hardware pattern directly inspired software pipeline designs.
  • 1973 -- Unix pipes: Doug McIlroy proposed connecting programs via standard streams. Ken Thompson implemented the | operator in Unix V3, establishing the "small programs connected by pipes" philosophy that defines Unix to this day.
  • 1990s -- ETL pipelines: The rise of data warehousing created demand for Extract-Transform-Load pipelines. Tools like Informatica and DataStage formalized multi-stage data processing.
  • 2000s -- CI/CD pipelines: Continuous integration servers (Jenkins, Hudson, CruiseControl) introduced build-test-deploy pipelines. The term "pipeline" became standard in DevOps vocabulary.
  • 2010s -- Stream processing: Apache Kafka, Spark Streaming, and Flink brought real-time pipeline processing to massive scale. The distinction between batch and stream pipelines began to blur.
  • 2015+ -- ML pipelines: TensorFlow Extended (TFX), Kubeflow, and MLflow introduced pipelines for machine learning: data ingestion, feature engineering, training, evaluation, and serving as connected stages.
  • 2020s -- Epistemic pipelines: The Prismatic Platform pioneered multi-level epistemic pipelines that process not just data but beliefs, confidence levels, and logical proofs through sequential reasoning gates.

Core Concepts

Stage and Transformation

Every pipeline consists of stages (also called steps, phases, or operators). Each stage implements a transformation function: it accepts input of type A, performs computation, and produces output of type B. The stage's contract is simple -- given valid input, produce valid output. This simplicity is what makes pipelines composable.

Data Flow and Backpressure

Data flows through a pipeline from source (producer) to sink (consumer). In synchronous pipelines, each stage blocks until its successor accepts the output. In asynchronous pipelines, stages may buffer data. Backpressure is the mechanism by which a slow consumer signals upstream stages to reduce their production rate, preventing memory exhaustion. GenStage in the Elixir ecosystem provides first-class backpressure support through demand-driven processing.

Fan-Out and Fan-In

Not all pipelines are strictly linear. Fan-out (scatter) sends a single input to multiple parallel stages, while fan-in (gather) collects outputs from multiple stages into one. These patterns enable parallelism: a pipeline might fan out to process records across multiple workers, then fan in to aggregate results.

Idempotency and Exactly-Once Processing

Production pipelines must handle failures gracefully. Idempotent stages can be safely retried without producing duplicate effects -- a critical property for fault tolerance. Combined with checkpointing (recording progress), idempotency enables exactly-once processing semantics even when individual stages fail and restart.

Types of Pipelines

Data Pipelines

Data pipelines move and transform data between systems. They range from simple batch jobs (nightly CSV imports) to real-time stream processing architectures handling millions of events per second. Key concerns include schema evolution, data quality validation, and exactly-once delivery guarantees.

ETL/ELT Pipelines

ETL (Extract-Transform-Load) pipelines are the classic data warehousing pattern: extract from source systems, transform into the target schema, then load into the warehouse. Modern architectures increasingly prefer ELT (Extract-Load-Transform), loading raw data first and transforming in the warehouse using SQL, leveraging the warehouse's compute power.

CI/CD Pipelines

CI/CD pipelines automate the software delivery process: build, test, analyze, package, deploy. Each stage acts as a quality gate -- if tests fail, the pipeline halts. The Prismatic Platform implements an 11-phase pre-commit pipeline that enforces compilation warnings, Credo analysis, forbidden pattern detection, template validation, and design consistency before any code reaches the repository.

ML Pipelines

Machine learning pipelines orchestrate the ML lifecycle: data ingestion, feature engineering, model training, hyperparameter tuning, evaluation, and serving. Tools like Kubeflow Pipelines, MLflow, and TFX provide frameworks for reproducible, versioned ML pipelines. The key challenge is experiment tracking -- understanding which combination of data, features, and hyperparameters produced a given model.

Epistemic Pipelines

Unique to the Prismatic Platform, epistemic pipelines process beliefs and claims through logical reasoning stages. The 16-level epistemic pipeline subjects every platform assertion to structural consistency checks (graph theory), logical consistency validation (rule-based), and formal necessity proofs (modal logic, Lean4). This Trinity Gate system ensures no claim is established without passing all three verification layers.

Technical Deep Dive

Pipeline Composition Strategies

Linear composition chains stages sequentially: A -> B -> C -> D. This is the simplest and most common pattern, exemplified by Unix pipes and the Elixir |> operator.

Directed Acyclic Graph (DAG) composition allows stages to have multiple inputs and outputs, forming a graph rather than a chain. Apache Airflow, Prefect, and Dagster model pipelines as DAGs, enabling parallel execution of independent stages while respecting dependencies.

Dynamic composition selects stages at runtime based on data content or configuration. A pipeline might route financial transactions through different validation stages depending on amount thresholds or originating country.

Pipeline Patterns

Several well-established patterns govern pipeline architecture:

  • Scatter-Gather: Fan out work to parallel processors, gather results. Used for parallel data processing and map-reduce operations.
  • Saga Pattern: A sequence of local transactions where each stage has a compensating action. If stage N fails, stages N-1 through 1 execute their compensations in reverse order, achieving eventual consistency.
  • Choreography: Each stage publishes events, and downstream stages subscribe. No central orchestrator -- the pipeline emerges from event subscriptions. Loose coupling but harder to observe end-to-end.
  • Orchestration: A central coordinator invokes each stage in sequence, handling errors and retries. Easier to reason about but introduces a single point of coordination.
  • Dead Letter Queue: Failed items are routed to a separate queue for inspection and manual or automated retry, preventing a single bad record from blocking the entire pipeline.

Backpressure Mechanisms

Without backpressure, a fast producer can overwhelm a slow consumer, causing unbounded memory growth. Backpressure strategies include:

  • Demand-driven (pull-based): Consumers request a specific number of items. GenStage uses this approach -- consumers send demand upstream, and producers only emit items when demand exists.
  • Rate limiting: Producers emit at a fixed maximum rate regardless of consumer capacity.
  • Buffering with overflow: Intermediate buffers absorb bursts, with overflow policies (drop oldest, drop newest, block producer) when buffers fill.
  • Credit-based flow control: Consumers issue "credits" to producers. Each emitted item consumes a credit. When credits are exhausted, the producer pauses.

Prismatic Platform Implementation

16-Level Epistemic Pipeline

The Prismatic Platform's epistemic pipeline processes every knowledge claim through 16 ordered levels of verification, implementing the NABLA Infinity framework. Each level adds confidence to or rejects assertions:

  1. Signal acquisition and plurality verification (minimum 2 independent sources)
  2. Contradiction detection and preservation (both sides maintained)
  3. Provenance chain validation (every belief traceable to origin)
  4. Time decay assessment (older evidence weighted less)
  5. Source independence verification
  6. Structural consistency check (belief graph forms valid DAG) 7-12. Progressive logical consistency validation 13-14. Formal necessity proofs (modal logic)
  7. Trinity Gate passage (all three layers must agree)
  8. Confidence threshold evaluation (0.95 for critical decisions)

SEADF Evolution Pipeline

The Scanner-Evolve-Analyze-Defend-Fix pipeline drives autonomous platform improvement. Each session triggers a pipeline that scans for quality issues, evolves solutions, analyzes their impact, defends against regressions, and fixes remaining problems. The pipeline is self-referential -- it improves its own detection capabilities across generations.

11-Phase Pre-Commit Pipeline

The platform's pre-commit hook implements an 11-phase quality gate pipeline:

  1. Compilation with --warnings-as-errors
  2. Credo strict analysis
  3. Dialyzer type checking
  4. Forbidden patterns scan (mocks, stubs, placeholders)
  5. Test execution for changed files
  6. Typespec coverage verification
  7. DateTime precision validation
  8. Template validation (promo site)
  9. Memory safety checks
  10. Design consistency validation
  11. Quality DNA update

Each phase is a hard gate -- failure at any phase blocks the commit entirely.

Quality Gates Pipeline

The platform's quality gates form a pipeline where each gate must pass before the next executes:

defmodule Mix.Tasks.Quality.Gates do
  @gates [
    {:compilation, &check_compilation/0},
    {:dialyzer, &check_dialyzer/0},
    {:credo, &check_credo/0},
    {:tests, &check_tests/0},
    {:coverage, &check_coverage/0},
    {:forbidden_patterns, &check_forbidden_patterns/0},
    {:typespec_coverage, &check_typespecs/0},
    {:impl_coverage, &check_impl_annotations/0}
  ]

  def run(_args) do
    results =
      Enum.reduce_while(@gates, [], fn {name, checker}, acc ->
        case checker.() do
          :ok -> {:cont, [{name, :passed} | acc]}
          {:error, reason} -> {:halt, [{name, {:failed, reason}} | acc]}
        end
      end)

    report_results(Enum.reverse(results))
  end
end

Code Examples

Elixir Pipe Operator

The |> operator is Elixir's syntactic pipeline primitive. It passes the result of the left expression as the first argument to the right function:

defmodule PrismaticPerimeter.AssetDiscovery do
  @moduledoc "Asset discovery pipeline using pipe operator composition."

  @spec discover(String.t()) :: {:ok, list(map())} | {:error, term()}
  def discover(domain) do
    domain
    |> normalize_domain()
    |> enumerate_subdomains()
    |> resolve_dns()
    |> fetch_certificates()
    |> scan_ports()
    |> assess_vulnerabilities()
    |> calculate_scores()
    |> persist_results()
  end

  @spec normalize_domain(String.t()) :: String.t()
  defp normalize_domain(domain) do
    domain
    |> String.downcase()
    |> String.trim()
    |> String.replace(~r/^https?:\/\//, "")
    |> String.replace(~r/\/.*$/, "")
  end

  @spec enumerate_subdomains(String.t()) :: [String.t()]
  defp enumerate_subdomains(domain) do
    [domain | SubdomainEnumerator.enumerate(domain)]
  end

  # Each subsequent function transforms the output of the previous
  # stage, maintaining a clear data flow through the pipeline.
end

GenStage Producer-Consumer Pipeline

GenStage provides demand-driven pipeline stages with automatic backpressure:

defmodule PrismaticOSINT.Pipeline.Producer do
  @moduledoc "Produces events from a data source with demand-driven flow control."
  use GenStage

  @impl GenStage
  def init(source) do
    {:producer, %{source: source, buffer: []}}
  end

  @impl GenStage
  def handle_demand(demand, state) when demand > 0 do
    {events, remaining} = Enum.split(state.buffer, demand)
    {:noreply, events, %{state | buffer: remaining}}
  end
end

defmodule PrismaticOSINT.Pipeline.Normalizer do
  @moduledoc "Normalizes raw events, filtering invalid entries."
  use GenStage

  @impl GenStage
  def init(:ok) do
    {:producer_consumer, :ok}
  end

  @impl GenStage
  def handle_events(events, _from, state) do
    normalized =
      events
      |> Enum.map(&normalize/1)
      |> Enum.reject(&is_nil/1)

    {:noreply, normalized, state}
  end

  defp normalize(%{type: :domain} = event) do
    %{event | value: String.downcase(event.value)}
  end

  defp normalize(%{type: :ip} = event) do
    case :inet.parse_address(to_charlist(event.value)) do
      {:ok, _} -> event
      {:error, _} -> nil
    end
  end
end

defmodule PrismaticOSINT.Pipeline.Scorer do
  @moduledoc "Terminal consumer that scores and persists events."
  use GenStage

  @impl GenStage
  def init(:ok), do: {:consumer, :ok}

  @impl GenStage
  def handle_events(events, _from, state) do
    Enum.each(events, fn event ->
      score = calculate_risk_score(event)
      PrismaticStorage.persist(%{event | score: score})
    end)

    {:noreply, [], state}
  end
end

# Pipeline topology: Producer -> Normalizer -> Scorer
defmodule PrismaticOSINT.Pipeline.Supervisor do
  use Supervisor

  @impl Supervisor
  def init(_arg) do
    children = [
      {PrismaticOSINT.Pipeline.Producer, :osint_feed},
      {PrismaticOSINT.Pipeline.Normalizer, :ok},
      {PrismaticOSINT.Pipeline.Scorer, :ok}
    ]

    Supervisor.init(children, strategy: :rest_for_one)
  end
end

Broadway Concurrent Data Pipeline

Broadway builds on GenStage to provide production-ready concurrent pipelines with batching, fault tolerance, and graceful shutdown:

defmodule PrismaticOSINT.Broadway.CertificateIngester do
  @moduledoc """
  Broadway pipeline for ingesting and persisting certificate data.
  Demonstrates batching, fault tolerance, and telemetry integration.
  """
  use Broadway

  @impl Broadway
  def handle_message(:default, message, _context) do
    certificate = parse_certificate(message.data)

    message
    |> Broadway.Message.update_data(fn _data -> certificate end)
    |> Broadway.Message.put_batcher(:persist)
  end

  @impl Broadway
  def handle_batch(:persist, messages, _batch_info, _context) do
    certificates = Enum.map(messages, & &1.data)

    case PrismaticStorage.Ecto.insert_all(Certificate, certificates) do
      {:ok, count} ->
        :telemetry.execute(
          [:prismatic, :osint, :certificates, :ingested],
          %{count: count},
          %{}
        )
        messages

      {:error, reason} ->
        Enum.map(messages, &Broadway.Message.failed(&1, reason))
    end
  end

  defp parse_certificate(raw_data) do
    raw_data
    |> Jason.decode!()
    |> Map.take(["subject", "issuer", "not_before", "not_after", "san"])
    |> normalize_certificate_fields()
  end
end

Best Practices

  1. Keep stages pure: Each pipeline stage should be a pure function where possible -- same input always produces same output, no side effects. Push side effects (database writes, API calls, logging) to the final stage or to dedicated side-effect stages.

  2. Design for composability: Stages should accept and return the same type (or a compatible type). This enables stages to be freely rearranged, inserted, or removed without changing adjacent stages.

  3. Use GenStage for backpressure: When processing data from external sources (APIs, message queues, file systems), use GenStage or Broadway to prevent overwhelming downstream stages. The demand-driven model ensures producers only generate data when consumers are ready.

  4. Batch for efficiency: When a pipeline terminates in a database write or API call, use Broadway's batching to group individual items into bulk operations. The Prismatic Platform batches certificate inserts into groups of 100 for 10x throughput improvement.

  5. Supervise pipeline processes: Use strategy: :rest_for_one for pipeline supervision so that when a middle stage crashes, all downstream stages are also restarted, preventing stale state or orphaned processes.

  6. Instrument with telemetry: Attach telemetry events at stage boundaries to measure throughput, latency, and error rates per stage. This identifies bottlenecks without adding debugging code.

  7. Make stages idempotent: Stages that can be safely retried simplify error recovery enormously. Use deterministic identifiers and upsert operations rather than blind inserts.

  8. Version your pipeline schemas: When input or output formats change, old and new stages must coexist during rollout. Use schema versioning and evolution strategies.

Anti-Patterns

  • The Mega-Stage: Cramming all logic into a single monolithic stage defeats the purpose of a pipeline. Each stage should be independently testable and replaceable. If a stage has multiple responsibilities, split it.

  • Silent Swallowing: Stages that catch exceptions and silently drop records make debugging impossible. Always log failures, route to dead letter queues, or propagate errors explicitly.

  • Unbounded Buffering: Placing unlimited queues between stages masks backpressure problems until the system runs out of memory under load. Always set buffer limits and define overflow policies.

  • Tight Coupling Between Stages: Stages that depend on internal implementation details of adjacent stages break when those stages change. Define clear contracts (input/output types) between stages.

  • No Observability: Pipelines without metrics are black boxes. When throughput drops or latency spikes, you need per-stage timing data to identify the bottleneck. Instrument from day one.

  • Ignoring Poison Pills: A single malformed record should not halt an entire pipeline. Implement poison pill detection -- after N retries, route the problematic record to a dead letter queue and continue processing.

  • Side Effects in Middle Stages: Database writes, API calls, or logging in middle pipeline stages makes them difficult to test, debug, and retry. Isolate side effects at pipeline boundaries.

Error Handling in Pipelines

Robust error handling distinguishes production pipelines from prototypes. Key strategies include:

Retry with exponential backoff: Transient failures (network timeouts, temporary unavailability) resolve themselves. Retry with increasing delays (1s, 2s, 4s, 8s) and a maximum retry count prevents both premature failure and infinite loops.

Circuit breakers: When a downstream system is consistently failing, a circuit breaker "opens" to stop sending requests, allowing the system to recover. After a cooldown period, the circuit breaker "half-opens" to test recovery before fully resuming traffic.

Dead letter queues: Records that fail after all retries are routed to a dedicated queue for manual inspection. This prevents a single bad record from blocking pipeline progress while preserving the record for later analysis.

Compensating transactions: In saga-style pipelines, each stage has an "undo" operation. When stage 5 of 7 fails, stages 4 through 1 execute their compensating transactions to maintain consistency.

Graceful degradation: When non-critical enrichment stages fail, the pipeline can skip them and proceed with reduced data quality rather than halting entirely. Mark records as partially processed for later backfill.

Pipeline Observability

Effective pipeline monitoring requires visibility into several dimensions:

  • Throughput: Records processed per second at each stage. Declining throughput at a specific stage reveals bottlenecks.
  • Latency: Time spent in each stage. P50, P95, and P99 latencies reveal different problems -- high P99 suggests occasional slow paths.
  • Queue depth: Number of items waiting between stages. Growing queues indicate a consumer cannot keep up with its producer.
  • Error rate: Percentage of records failing at each stage. Sudden spikes correlate with upstream data quality changes or downstream system failures.
  • Backpressure metrics: Demand satisfaction rate in GenStage pipelines. If consumers consistently demand more than producers can supply, the pipeline is producer-bound.

The Prismatic Platform emits telemetry events from every pipeline stage, enabling real-time dashboards and alerting through the quality monitoring infrastructure.

TechnologyDescriptionPipeline Role
Apache KafkaDistributed event streaming platformInter-stage buffer and pipeline backbone
Apache AirflowDAG-based workflow orchestratorBatch pipeline scheduling and monitoring
Apache BeamUnified batch/stream processing modelWrite-once, run-anywhere pipeline logic
GenStageElixir demand-driven processing libraryBackpressure-aware pipeline stages
BroadwayElixir production data pipeline libraryConcurrent processing with batching and fault tolerance
DagsterModern data pipeline orchestratorType-checked, testable pipeline definitions
PrefectPython-native workflow orchestrationDynamic pipeline composition with retries
LuigiPython batch pipeline framework (Spotify)Dependency resolution for batch jobs

Future Directions

Pipeline architectures continue to evolve in several directions:

  • Unified batch and stream: The distinction between batch and streaming pipelines is dissolving. Frameworks like Apache Beam and Flink provide unified APIs that work for both modes.
  • AI-driven pipeline optimization: Machine learning models that predict optimal parallelism, batch sizes, and resource allocation for pipeline stages based on historical performance data.
  • Self-healing pipelines: Pipelines that automatically detect degraded stages, reroute traffic, scale resources, and alert operators -- moving from reactive to proactive fault management.
  • Declarative pipeline definition: Specifying what the pipeline should achieve rather than how to process data. The runtime optimizes stage ordering, parallelism, and resource allocation automatically.
  • Epistemic pipeline standardization: The Prismatic Platform's approach to multi-level belief verification may establish patterns for other AI-safety systems requiring rigorous evidence processing.

See Also

  • Data Pipeline -- specialized pipelines for moving and transforming data between systems
  • CI/CD -- continuous integration and delivery pipelines for software deployment
  • ETL -- Extract-Transform-Load pipelines for data warehousing
  • Elixir -- functional language with first-class pipeline support via the |> operator
  • GenStage -- Elixir library for demand-driven pipeline stages
  • Broadway -- production-ready concurrent data pipelines in Elixir
  • Quality Gates -- pipeline checkpoints that enforce quality thresholds
  • Workflow -- broader execution patterns that may include pipeline stages
  • Stream Processing -- real-time data processing architectures
  • Circuit Breaker -- fault tolerance pattern critical for pipeline resilience
  • Actor Model -- concurrency model underlying Elixir's pipeline implementations
  • Static Analysis -- automated code analysis often integrated into CI/CD pipelines
  • Adapter Pattern -- pattern for normalizing different data sources into pipeline-compatible formats

Connect & Contribute

Created by Tomas Korcak (korczis) | Open Source under GHL

Back to Glossary