Skip to main content

Pipelines & Jobs

A pipeline defines how data flows from inputs, through processors, to outputs. A job is how you deploy a pipeline to Expanso Cloud.

Pipeline Structure​

Every pipeline has three main sections:

input:
  # Where data comes from
  kafka:
    addresses: ["localhost:9092"]
    topics: ["logs"]

pipeline:
  processors:
    # What to do with the data
    - mapping: |
        root.message = this.msg.uppercase()
        root.timestamp = now()

output:
  # Where data goes
  aws_s3:
    bucket: processed-logs
    path: "logs/${!timestamp_unix()}.json"

Inputs​

Sources of data:

  • Message queues: Kafka, RabbitMQ, NATS
  • Databases: PostgreSQL, MySQL, MongoDB (nightly only)
  • Files: Local, S3, SFTP
  • HTTP: Webhooks, APIs
  • Streams: TCP, UDP, WebSocket
  • Generate: Test data

As of September 10, 2026, MongoDB input, output, processor, and cache components are available in nightly v2.1.22-nightly.134.g0858e17a. They are not included in stable v2.1.21, which the default installer selects. MongoDB pipelines require an edge-node release that includes these components.

Browse all inputs →

Processors​

Transform, filter, and enrich data:

  • Mapping: Transform with Bloblang
  • Filter: Drop unwanted messages
  • Parse: JSON, CSV, XML, Avro, Protobuf
  • Enrich: Lookup external data
  • Aggregate: Batch, window, group

Browse all processors →

Outputs​

Destinations for processed data:

  • Message queues: Kafka, RabbitMQ, NATS
  • Databases: PostgreSQL, Elasticsearch
  • Object storage: S3, GCS, Azure Blob
  • HTTP: Webhooks, APIs
  • Files: Local, S3, SFTP
  • Observability: Datadog, Splunk, Prometheus, and similar via the HTTP output

Browse all outputs →

Simple Example​

Read files, filter, output to terminal:

input:
  file:
    paths: ["/var/log/*.log"]

pipeline:
  processors:
    - mapping: |
        # Only keep ERROR logs
        root = if !this.contains("ERROR") { deleted() }

output:
  stdout:
    codec: lines

Multi-Output Example​

Route data to different destinations:

input:
  http_server:
    address: "0.0.0.0:8080"

pipeline:
  processors:
    - mapping: |
        root = this.parse_json()

output:
  broker:
    pattern: fan_out
    outputs:
      # Errors to Slack
      - switch:
          cases:
            - check: this.level == "ERROR"
              output:
                http_client:
                  url: "https://slack.com/webhook"

      # Metrics to a Kafka topic
      - switch:
          cases:
            - check: this.type == "metric"
              output:
                kafka:
                  addresses: ["localhost:9092"]
                  topic: metrics

      # Everything to S3
      - aws_s3:
          bucket: all-events

Pipeline Config vs Job Format​

There are two YAML formats and it's important to know when to use each:

Pipeline config format (the processing logic)​

The pipeline itself, with input/pipeline/output at the root level. This is what expanso-edge validate checks, and what gets nested inside a job. It is not something you run directly:

input:
  file:
    paths: ["./data.json"]
pipeline:
  processors:
    - mapping: "root = this"
output:
  stdout: {}

Job format (what you deploy)​

Used with expanso-cli job deploy, for Expanso Cloud and for a local agent alike. Wraps the pipeline config under a config: key with name and type at the top level:

name: my-file-processor
type: pipeline
config:
  input:
    file:
      paths: ["./data.json"]
  pipeline:
    processors:
      - mapping: "root = this"
  output:
    stdout: {}
Converting Between Formats

To convert a pipeline config to job format: add name and type: pipeline at the top level, then indent the entire pipeline config under config:. See Deploy to Cloud for the full walkthrough.

Deployment​

Pipelines deploy to nodes based on:

  • Direct selection: Choose specific nodes
  • Label selectors: Target nodes with matching labels
  • Workspace: All nodes in a workspace

Example with labels:

# Deploy to production log processors only
selector:
  env: production
  role: log-processor

What's Next?​

👉 Components - Learn about building blocks

👉 Build a Pipeline - Get hands-on

👉 Bloblang Guide - Master data transformations