Skip to main content

Complete Pipeline

Copy pipeline YAML
Review it for your environment before you deploy.

This HTTP reference converts JSON, CSV, and XML. It detects the source from the payload, chooses the response format from the Accept header, and returns the result to the originating request.

Full Configuration​

Complete pipeline
transform-formats.yaml
name: transform-formats-complete
type: pipeline
description: Complete Format Transformation Pipeline - Transforms between JSON, CSV, and XML formats with auto-detection.
namespace: production
labels:
  category: data-transformation
  pattern: format-transformation

config:
  input:
    http_server:
      address: '0.0.0.0:8080'
      path: /transform
      allowed_verbs: ['POST']
      sync_response:
        headers:
          Content-Type: '${! meta("response_content_type") }'

  pipeline:
    processors:
      # Step 1: Detect input format from Content-Type or structure
      - mapping: |
          let content = content().string()
          meta http_accept = meta("Accept").or("application/json")

          root.source_format = if $content.has_prefix("{") || $content.has_prefix("[") {
            "json"
          } else if $content.has_prefix("<?xml") || $content.has_prefix("<") {
            "xml"
          } else if $content.split(",").length() - 1 >= 1 {
            "csv"
          } else {
            "text"
          }

          root.raw = $content
          meta source_format = root.source_format

      # Step 2: Parse to normalized structure
      - switch:
          - check: 'meta("source_format") == "json"'
            processors:
              - mapping: root = this.raw.parse_json()
          - check: 'meta("source_format") == "xml"'
            processors:
              - mapping: root = this.raw
              - xml:
                  operator: to_json

          - check: 'meta("source_format") == "csv"'
            processors:
              - mapping: root = this.raw.parse_csv()

      # Step 3: Transform to target format (based on Accept header or default)
      - mapping: |
          root = this
          let target = meta("http_accept").or("application/json")

          meta target_format = if $target.contains("xml") {
            "xml"
          } else if $target.contains("csv") {
            "csv"
          } else {
            "json"
          }

          meta response_content_type = match meta("target_format") {
            "xml" => "application/xml"
            "csv" => "text/csv"
            _ => "application/json"
          }

      - mapping: |
          root.data = this
          root.transformation = {
            "source_format": meta("source_format"),
            "target_format": meta("target_format"),
            "transformed_at": now()
          }

      - switch:
          - check: 'meta("target_format") == "xml"'
            processors:
              - mapping: root = this.format_xml()
          - check: 'meta("target_format") == "csv"'
            processors:
              - mapping: |
                  let rows = if this.data.type() == "array" { this.data } else { [this.data] }
                  let columns = $rows.map_each(row -> row.keys()).flatten().unique().sort()
                  let header = $columns.map_each(col -> "\"" + col.replace_all("\"", "\"\"") + "\"").join(",")
                  let lines = $rows.map_each(row -> $columns.map_each(col -> "\"" + row.get(col).or("").string().replace_all("\"", "\"\"") + "\"").join(","))
                  root = $header + "\r\n" + $lines.join("\r\n") + "\r\n"

  output:
    # Reply to the HTTP request that delivered the message
    sync_response: {}

logger:
  level: INFO
  format: json

metrics:
  prometheus:
    path: /metrics

Quick Test​

# Request CSV output from JSON input
curl -X POST http://localhost:8080/transform \
  -H "Content-Type: application/json" \
  -H "Accept: text/csv" \
  -d '{
    "sensor_id": "temp-001",
    "temperature": 72.5,
    "humidity": 45.0,
    "timestamp": "2024-01-15T10:30:00Z"
  }'

# Response contains the CSV header and values

Deploy​

# Deploy to Expanso orchestrator
expanso-cli job deploy transform-formats.yaml

# Or run locally with expanso-edge
expanso-edge run --config transform-formats.yaml

Download​

The download is a separate JSON-to-binary-Avro integration variant at /sensors/ingest. It requires Node.js on every edge node and sends records to Kafka. It does not execute the JSON/CSV/XML response example above.

Download transform-formats.yaml

What's Next?​