Skip to main content

Metadata Enrichment for Lineage

This example demonstrates how to stamp every message with pipeline provenance, node identity, and custom compliance tags using the metadata processor. Use it as a starting point for building OpenLineage-style lineage events or compliance audit trails.

Download & Run​

Quick Start:

# 1. Download the pipeline and check it offline
curl -sSLo pipeline.yaml \
  https://docs.expanso.io/examples/metadata-enrichment.yaml
expanso-edge validate pipeline.yaml

# 2. Start a local edge agent (its state stays in ./.edge)
expanso-edge run --local --data-dir ./.edge

# 3. In a second terminal, deploy the pipeline to it as a job
export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli"
{ printf 'name: metadata-enrichment\ntype: pipeline\nconfig:\n'
  sed 's/^/  /' pipeline.yaml; } |
  expanso-cli --endpoint http://localhost:9010 job deploy -

This runs on your machine with no account. Step 3 wraps the pipeline in the name, type and config fields that a job needs, and a stdout output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See Local Mode for more, or Deploy to Cloud to run it on managed nodes.

Download: metadata-enrichment.yaml

What This Pipeline Does​

  1. Generates a synthetic event every 5 seconds (no external dependencies)
  2. Enriches each event with pipeline, node, and OpenLineage-aligned identity fields
  3. Tags events with custom compliance metadata (owner, compliance tier)
  4. Outputs the enriched JSON to stdout so you can inspect the result

Complete Pipeline​

input:
  generate:
    mapping: |
      root = {
        "event_id": uuid_v4(),
        "level": "info",
        "message": "user signup"
      }
    interval: 5s

pipeline:
  processors:
    - metadata:
        include: [core, pipeline, node]
        custom:
          pipeline_owner: [email protected]
          compliance_tier: pii
        target: body
        format: nested
        body_key: lineage

output:
  stdout:
    codec: lines

Try Locally​

This pipeline needs nothing external, so it runs on a local agent with no account. Start one as in Step 3 of Try It Without an Account, then run this in a second terminal, from the same folder:

curl -fsSL -o metadata-enrichment.yaml \
  https://docs.expanso.io/examples/metadata-enrichment.yaml
expanso-edge validate metadata-enrichment.yaml
{ printf 'name: metadata-enrichment\ntype: pipeline\nconfig:\n'
  sed 's/^/  /' metadata-enrichment.yaml; } > metadata-job.yaml
export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli"
expanso-cli job deploy metadata-job.yaml \
  --endpoint http://localhost:9010

The printf and sed lines wrap the pipeline in the job fields (name, type, config) that job deploy expects. EXPANSO_CLI_CONFIG_DIR and --endpoint keep the CLI on the local agent, away from any saved profile or Expanso Cloud workspace. Every 5 seconds an event appears in the agent's terminal as one JSON line. Formatted for reading, it looks like this (IDs and times vary):

{
  "event_id": "...",
  "level": "info",
  "lineage": {
    "agent_version": "",
    "cluster_name": "",
    "compliance_tier": "pii",
    "environment": "",
    "event_time": "...",
    "git_branch": "",
    "git_commit_sha": "",
    "git_repo_url": "",
    "hostname": "",
    "job_name": "metadata-enrichment",
    "job_namespace": "default",
    "node_id": "...",
    "pipeline_name": "metadata-enrichment",
    "pipeline_owner": "[email protected]",
    "pipeline_version": 1,
    "producer": "https://expanso.io/edge",
    "region": "",
    "run_id": "e-..."
  },
  "message": "user signup"
}

On a local agent, the node fields such as hostname, region, and agent_version are empty, and node_id is derived from your machine's host name. The job keeps running until you stop it:

expanso-cli job stop metadata-enrichment --force \
  --endpoint http://localhost:9010

On a local agent the pipeline can keep producing output for a short while after the stop (22 seconds in one test), then halts. Stopping the agent with Ctrl+C stops everything at once.

Configuration Breakdown​

Input: Generate​

input:
  generate:
    mapping: |
      root = {
        "event_id": uuid_v4(),
        "level": "info",
        "message": "user signup"
      }
    interval: 5s

The generate input emits a synthetic message every 5 seconds, so the pipeline runs anywhere without external dependencies. Replace it with kafka, file, or any other input when wiring up to real data.

See: Generate Input

Processor: Metadata Enrichment​

- metadata:
    include: [core, pipeline, node]
    custom:
      pipeline_owner: [email protected]
      compliance_tier: pii
    target: body
    format: nested
    body_key: lineage

The metadata processor splices three categories of runtime metadata under a lineage key:

  • core — OpenLineage-aligned identity (run_id, job_name, job_namespace, event_time, producer)
  • pipeline — pipeline name and version
  • node — node ID, region, environment, cluster, hostname, agent version

The custom: block adds two static fields that travel with every event for ownership and compliance tracking.

See: Metadata Processor Reference

Output: Stdout​

output:
  stdout:
    codec: lines

Each enriched event is printed as a single JSON line. Swap this for kafka, aws_s3, or http_client when you're ready to send the data downstream.

Example Output​

A single emitted event looks like this:

{
  "event_id": "9f3e1b8a-2c5d-4e7f-9a01-3b2c4d5e6f7a",
  "level": "info",
  "message": "user signup",
  "lineage": {
    "run_id": "01HF7K8M9N0P1Q2R3S4T5V6W7X",
    "job_name": "lineage-demo",
    "job_namespace": "default",
    "event_time": "2026-04-30T12:34:56.789Z",
    "producer": "https://expanso.io/edge/v1.4.2",
    "pipeline_name": "lineage-demo",
    "pipeline_version": "v1",
    "git_commit_sha": "",
    "git_repo_url": "",
    "git_branch": "",
    "node_id": "node-eu-west-1-a",
    "hostname": "edge-01",
    "region": "eu-west-1",
    "environment": "production",
    "cluster_name": "edge-eu",
    "agent_version": "1.4.2",
    "pipeline_owner": "[email protected]",
    "compliance_tier": "pii"
  }
}

Common Variations​

Attach to message metadata instead of the body

Use target: meta to write to message metadata, then reference fields downstream via interpolation (${! metadata("region") }).

- metadata:
    include: [core, pipeline, node]
    target: meta

Surface runtime counters

Opt in to the runtime category to attach records_in, bytes_out, error_count, and duration_ms:

- metadata:
    include: [pipeline, runtime]
    target: body
    format: nested
    body_key: stats

Flat-merge into the body

If your downstream consumer expects a flat JSON object, use format: flat. Note that metadata keys overwrite body keys on collision — use exclude to drop fields that would clash.

- metadata:
    include: [pipeline]
    exclude: [git_commit_sha, git_repo_url, git_branch]
    target: body
    format: flat

Promote inbound metadata (e.g. Sparkplug B) into the body

For SCADA / IIoT inputs that stamp domain metadata on every message (e.g. Sparkplug B's spb_* keys, Kafka headers, MQTT topic parts) and emit a scalar metric value as the body, use the promote block to land those metadata keys as columns without per-pipeline Bloblang glue:

- metadata:
    target: body
    format: flat
    promote:
      match: "spb_"
      rename: { spb_metric_name: metric, spb_datatype: datatype }
      cast:   { spb_timestamp: int, spb_is_historical: bool }
      wrap_key: value          # wraps a scalar body so promoted fields can attach

See the dedicated Promote Inbound Metadata example for the full SCADA-shaped walkthrough, and Promoting inbound metadata into the body for the field reference.

Next Steps​