Benthos Stream Processing Pipeline

input:
  kafka:
    addresses: ["kafka:9092"]
    topics: ["order-events"]
    consumer_group: benthos-pipeline

pipeline:
  processors:
    - mapping: |
        root = this
        root.processed_at = now()
        root.total = this.quantity * this.price

output:
  s3:
    bucket: processed-orders
    path: "orders/${!timestamp_unix()}.json"
    region: eu-west-1

# Kafka
input:
  kafka:
    addresses: ["kafka:9092"]
    topics: ["events"]

# HTTP server (receive webhooks)
input:
  http_server:
    path: /webhook
    allowed_verbs: ["POST"]

# S3 (process new files)
input:
  aws_s3:
    bucket: raw-data
    prefix: uploads/
    region: eu-west-1

# Database polling
input:
  sql_select:
    driver: postgres
    dsn: "postgres://user:pass@db:5432/app"
    table: events
    columns: ["id", "data", "created_at"]
    where: "processed = false"

# AMQP (RabbitMQ)
input:
  amqp_0_9:
    urls: ["amqp://rabbitmq:5672"]
    queue: tasks

pipeline:
  processors:
    # Transform fields
    - mapping: |
        root.user_id = this.user.id
        root.email = this.user.email.lowercase()
        root.order_total = this.items.map_each(i -> i.price * i.qty).sum()
        root.currency = "EUR"
        root.timestamp = this.created_at.ts_parse("2006-01-02T15:04:05Z")

    # Filter messages
    - mapping: |
        root = if this.order_total < 10 { deleted() }

    # Enrich with HTTP call
    - branch:
        request_map: |
          root.url = "http://user-api:8080/users/" + this.user_id
        processors:
          - http:
              url: "${! this.url }"
              verb: GET
        result_map: |
          root.user_name = this.name
          root.user_tier = this.tier

# Multiple outputs (fan-out)
output:
  broker:
    outputs:
      - s3:
          bucket: order-archive
          path: "${!meta(\"kafka_topic\")}/${!count(\"files\")}.json"

      - kafka:
          addresses: ["kafka:9092"]
          topic: enriched-orders

      - http_client:
          url: https://analytics.myorg.com/events
          verb: POST
          headers:
            Authorization: "Bearer ${API_TOKEN}"

      - sql_insert:
          driver: postgres
          dsn: "postgres://user:pass@db:5432/analytics"
          table: orders
          columns: ["user_id", "total", "timestamp"]
          args_mapping: |
            root = [this.user_id, this.order_total, this.timestamp]

output:
  broker:
    outputs:
      - kafka:
          addresses: ["kafka:9092"]
          topic: processed-orders
      - fallback:
          - kafka:
              addresses: ["kafka:9092"]
              topic: dead-letter-queue
          - file:
              path: /var/log/failed-messages.jsonl

input:
  kafka:
    addresses: ["kafka:9092"]
    topics: ["events"]
    batching:
      count: 100
      period: 5s

# docker-compose.yml
services:
  benthos:
    image: ghcr.io/redpandadata/connect
    volumes:
      - ./pipeline.yaml:/benthos.yaml
    environment:
      - API_TOKEN=${API_TOKEN}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-pipeline
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: benthos
          image: ghcr.io/redpandadata/connect
          args: ["-c", "/config/pipeline.yaml"]
          volumeMounts:
            - name: config
              mountPath: /config
      volumes:
        - name: config
          configMap:
            name: benthos-pipeline