OpenLineage and OpenMetadata

Applies to VAST Cluster 5.5.

If your organization tracks data lineage, in Marquez, OpenMetadata or another tool that accepts OpenLineage events, you'll want each DataEngine run to show up there: this function read that object, ran at this time, and succeeded or failed. This page shows how a function emits OpenLineage run events with the openlineage-python client, over HTTP to a lineage backend or through a VAST Event Broker topic, and what that means for OpenMetadata.

This isn't a built-in DataEngine integration in 5.5. DataEngine doesn't emit lineage events, and has no OpenLineage, Marquez or OpenMetadata settings. Your handler emits the events, with a client you add to its dependencies, to a backend its pods can reach.

Commands use the environment variables from the DataEngine quickstart. The outputs come from a lab cluster running VAST Cluster 5.5.0-SP6 with vastde v5.5.0-sp2 and openlineage-python 1.53.0, where the registry host is 127.0.0.1:5000 and the trigger's source bucket is docs-src. Your names, GUIDs, addresses and times will differ.

What a function can reach

A function sends lineage events from its pods on the compute cluster, so the lineage backend, or the Event Broker, must be reachable from there. On the lab's compute cluster, a function could open connections to:

Destination	Result
An HTTP server on a host outside the compute cluster	Reached
The internet (`https://pypi.org`)	Reached
The VAST cluster's S3 endpoint	Reached
The VAST Event Broker's Kafka port (9092)	Reached

The tenant's namespace had one network policy, which limits only the traffic coming in to the function pods; it didn't restrict outgoing traffic. What your functions can reach depends on your compute cluster and your site's network: firewalls, routing and any policies your platform team adds. Ask them to confirm a route from the compute cluster to your lineage backend, or to the Event Broker, before you start.

Name the job, the run and the datasets

OpenLineage describes each run of a job, and the datasets it read and wrote. For a DataEngine function, the example uses:

OpenLineage	Value in the example	Where it comes from
Job namespace	`vast-dataengine://my-tenant`	Your choice. OpenLineage leaves the job namespace to you; the example uses one per tenant, from an environment variable `LINEAGE_NAMESPACE` in the pipeline.
Job name	`lineage-demo.lineagefn`	`<pipeline>.<function deployment>`, from the environment variables `VASTPIPELINE_PIPELINE_NAME` and `VASTPIPELINE_FUNCTION_DEPLOYMENT_NAME`, which DataEngine sets in every function container. They aren't documented.
Run ID	`00005015-0000-0016-2e04-8a6dbdde4eb9`	The event's DataEngine trace ID as a UUID, so one lineage run maps to one DataEngine trace. Read it from the current span; see get the trace ID.
Input dataset	namespace `s3://docs-src`, name `lineage-src/orders.bad`	The event's bucket and object key (`event.bucket`, `event.object_key`)

Don't use ctx.function_name for the job name: it's the generated service name of the function deployment, not your function's name, and it changes when the pipeline is re-created (see Context).

The example has no output datasets. For a function that writes to VAST DataBase, name the table the way the tool that reads it does. If you query it through Trino, OpenLineage's naming convention for Trino is namespace trino://<host>:<port> and name <catalog>.<schema>.<table> (OpenLineage naming), which lets the backend join your function's lineage with Trino's.

Write the function

The function builds the OpenLineage client and the job once in init, then emits START when an event arrives, and COMPLETE or FAIL when it's handled. A FAIL event carries the error message, and the handler re-raises the exception so DataEngine still records the run as failed. Objects whose key ends in .bad fail on purpose. Save this as lineagefn/main.py:

import os
import uuid
from datetime import datetime, timezone

from openlineage.client import OpenLineageClient
from openlineage.client.event_v2 import InputDataset, Job, Run, RunEvent, RunState
from openlineage.client.facet_v2 import error_message_run, job_type_job
from opentelemetry import trace

PRODUCER = "https://example.com/lineage-demo"  # identifies the code that emits the events
client = None
job = None

def init(ctx):
    global client, job
    client = OpenLineageClient()  # transport from OPENLINEAGE_URL
    job = Job(
        namespace=os.environ["LINEAGE_NAMESPACE"],
        name=f"{os.environ['VASTPIPELINE_PIPELINE_NAME']}.{os.environ['VASTPIPELINE_FUNCTION_DEPLOYMENT_NAME']}",
        facets={"jobType": job_type_job.JobTypeJobFacet(
            processingType="STREAMING", integration="VAST_DATAENGINE", jobType="FUNCTION")},
    )
    ctx.logger.info("lineage job %s/%s", job.namespace, job.name)

def emit(state, run, inputs):
    client.emit(RunEvent(eventType=state, eventTime=datetime.now(timezone.utc).isoformat(),
                         run=run, job=job, producer=PRODUCER, inputs=inputs))

def handler(ctx, event):
    # Run id = the DataEngine trace id as a UUID, so a lineage run maps to one DataEngine trace
    trace_id = trace.get_current_span().get_span_context().trace_id
    run = Run(runId=str(uuid.UUID(int=trace_id)))
    inputs = [InputDataset(namespace=f"s3://{event.bucket}", name=event.object_key)]
    emit(RunState.START, run, inputs)
    try:
        ctx.logger.info("processing %s, lineage run %s, trace %032x", event.object_key, run.runId, trace_id)
        if event.object_key.endswith(".bad"):
            raise ValueError(f"cannot process {event.object_key}")
    except Exception as err:
        failed = Run(runId=run.runId, facets={"errorMessage": error_message_run.ErrorMessageRunFacet(
            message=str(err), programmingLanguage="python")})
        emit(RunState.FAIL, failed, inputs)
        raise
    emit(RunState.COMPLETE, run, inputs)

client.emit sends each event from inside the handler, so the time it takes is part of the event's run. Decide what should happen to your pipeline when the lineage backend is down or slow. If an exception escapes emit, the event fails like any other; if you don't want lineage problems to fail events, catch and log errors around emit.

lineagefn/requirements.txt holds the client:

openlineage-python==1.53.0

The client's dependencies don't conflict with the packages the DataEngine runtime pins. Build with --save-build-log to see what the builder installed:

vastde functions build "$REGISTRY/docs/lineagefn" --target ./lineagefn --image-tag v1 --push --save-build-log

Detected language: python
Validating Python version 3.12.*...
Python version 3.12.* resolved to 3.12.13
Building 127.0.0.1:5000/docs/lineagefn:v1
App Path: ./lineagefn
Handlers File: main.py
2026/10/08 14:25:43 [Started] Python Builder: 127.0.0.1:5000/docs/lineagefn:v1
2026/10/08 14:26:31 [Completed] Python Builder: 127.0.0.1:5000/docs/lineagefn:v1
Build log saved to: lineagefn/build_lineagefn_v1_2026-10-08-14-25-42.log
Build completed: 127.0.0.1:5000/docs/lineagefn:v1

grep -o 'Successfully installed .*' lineagefn/build_lineagefn_v1_*.log | head -1 | tr ' ' '\n' | grep -E '^(openlineage|httpx|attrs|requests|pyyaml|PyYAML|packaging|python-dateutil)'

attrs-26.1.0
httpx2-2.13.1
openlineage-python-1.53.0
packaging-26.3
python-dateutil-2.9.0.post0
pyyaml-6.0.3
requests-2.34.2

vastde functions create --name lineagefn --artifact-type image --artifact-source docs/lineagefn --image-tag v1 --container-registry "$REGISTRY_NAME" --publish

Function created: lineagefn
Name: lineagefn
Tags: []
GUID: 7c66f295-a169-4d14-b2a3-3096afd91932
Owner: [id: 4294967295, id-type: vid]
Created At: 2026-10-08T14:26:44Z
Updated At: 2026-10-08T14:26:44Z
VRN: vast:dataengine:functions:lineagefn
Last Revision: 1

vastde triggers create element --name lineage-uploads --source-bucket "$BUCKET" --event "ObjectCreated:*" --name-prefix lineage-src/

Trigger created: lineage-uploads
Name: lineage-uploads
Description: N/A
Tags: []
GUID: e29f7f51-a869-4483-946e-f0f29b714999
Owner: {4294967295 vid 0xc000992620 map[]}
CreatedAt: 2026-10-08T14:26:55Z
UpdatedAt: 2026-10-08T14:26:55Z
VRN: vast:dataengine:triggers:lineage-uploads
Type: Element
Events:
  -: ObjectCreated:*
Topic: vast:dataengine:topics:docs-kafka/default
SourceBucket: docs-src

Send events over HTTP

OpenLineageClient() with no arguments takes its transport from environment variables: OPENLINEAGE_URL alone selects the HTTP transport, which posts each event to <url>/api/v1/lineage, the path Marquez serves. Set it, and the job namespace, in the pipeline. The link sets retries: 0, so the failing object fails once instead of being retried. Save this as lineage-demo.yaml, with your compute cluster, namespace and backend address:

name: lineage-demo
description: emits OpenLineage run events for every object
kubernetes_cluster_vrn: vast:dataengine:kubernetes-clusters:test-cluster
namespace: dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91
manifest:
  triggers:
    - name: uploads
      vrn: vast:dataengine:triggers:lineage-uploads
  function_deployments:
    - name: lineagefn
      function_vrn: vast:dataengine:functions:lineagefn
      config:
        environment_variables:
          - name: OPENLINEAGE_URL
            value: http://<lineage host>:15000
          - name: LINEAGE_NAMESPACE
            value: vast-dataengine://my-tenant
  links:
    - source: [uploads]
      destination: [lineagefn]
      config:
        retries: 0

Names that start with OPENLINEAGE aren't reserved, so the pipeline accepts them (see what the cluster rejects). If your backend needs an API key, put it in a secret, not an environment variable.

vastde pipelines create --config @lineage-demo.yaml --deploy

Pipeline created: lineage-demo
Name: lineage-demo
Description: emits OpenLineage run events for every object
Tags: []
GUID: 158b72ff-ba03-49d9-8feb-fb070173fc6a
Owner: [id: 4294967295, id-type: vid]
Created At: 2026-10-08T14:27:03Z
Updated At: 2026-10-08T14:27:03Z
VRN: vast:dataengine:pipelines:lineage-demo

Deploying pipeline: lineage-demo
Pipeline deployed successfully

until s=$(vastde pipelines get lineage-demo -o json | jq -r .status); [ "$s" = Ready ] || [ "$s" = Error ]; do sleep 5; done; vastde pipelines get lineage-demo -o json | jq -c "{status, reason}"

{"status":"Ready","reason":"Pipeline is ready and active."}

Upload one object that succeeds and one that fails:

echo lineage > orders.csv; aws --region us-east-1 --endpoint-url "$S3_ENDPOINT" s3 cp orders.csv "s3://$BUCKET/lineage-src/orders.csv" >/dev/null
cp orders.csv orders.bad; aws --region us-east-1 --endpoint-url "$S3_ENDPOINT" s3 cp orders.bad "s3://$BUCKET/lineage-src/orders.bad" >/dev/null
sleep 15; vastde logs get lineage-demo --since 3m --scope user

2026-10-08 14:27:35.67 [lineagefn] [INFO]  [user] openlineage.client.client: OpenLineageClient will use `http` transport
2026-10-08 14:27:35.67 [lineagefn] [INFO]  [user] lineage job vast-dataengine://my-tenant/lineage-demo.lineagefn
2026-10-08 14:28:04.86 [lineagefn] [INFO]  [user] processing lineage-src/orders.bad, lineage run 00005015-0000-0016-2e04-8a6dbdde4eb9, trace 00005015000000162e048a6dbdde4eb9
2026-10-08 14:28:05.48 [lineagefn] [INFO]  [user] processing lineage-src/orders.csv, lineage run 00005005-0000-001d-2e04-8a6dbdde4eb9, trace 000050050000001d2e048a6dbdde4eb9

vastde traces list lineage-demo --since 5m

Trace ID                          Status  Pipeline              Function              Start Time               End Time               
--------------------------------------------------------------------------------------------------------------------------------------
e36fcb7b93d53b31939c24747629f361  OK      lineage-demo          lineagefn             2026-10-08 14:27:35.669  2026-10-08 14:27:35.672
00005015000000162e048a6dbdde4eb9  ERROR   lineage-demo          lineagefn             2026-10-08 14:28:04.857  2026-10-08 14:28:04.867
000050050000001d2e048a6dbdde4eb9  OK      lineage-demo          lineagefn             2026-10-08 14:28:05.482  2026-10-08 14:28:05.488

Each lineage run ID is its DataEngine trace ID, written as a UUID: run 00005015-0000-0016-2e04-8a6dbdde4eb9 is trace 00005015000000162e048a6dbdde4eb9, the one that failed.

For this page the backend was a small HTTP server that printed each request it received, standing in for Marquez; Marquez itself wasn't run. It received four events, START and FAIL for orders.bad and START and COMPLETE for orders.csv. The first two:

{
  "from": "<pod address>",
  "path": "/api/v1/lineage",
  "content_type": "application/json",
  "body": {
    "eventTime": "2026-10-08T14:28:04.858094+00:00",
    "eventType": "START",
    "inputs": [
      {
        "facets": {},
        "inputFacets": {},
        "name": "lineage-src/orders.bad",
        "namespace": "s3://docs-src"
      }
    ],
    "job": {
      "facets": {
        "jobType": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.53.0/client/python",
          "_schemaURL": "https://openlineage.io/spec/facets/2-0-4/JobTypeJobFacet.json#/$defs/JobTypeJobFacet",
          "integration": "VAST_DATAENGINE",
          "jobType": "FUNCTION",
          "processingType": "STREAMING"
        }
      },
      "name": "lineage-demo.lineagefn",
      "namespace": "vast-dataengine://my-tenant"
    },
    "outputs": [],
    "producer": "https://example.com/lineage-demo",
    "run": {
      "facets": {
        "tags": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.53.0/client/python",
          "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/TagsRunFacet.json#/$defs/TagsRunFacet",
          "tags": [
            {
              "key": "openlineage_client_version",
              "source": "OPENLINEAGE_CLIENT",
              "value": "1.53.0"
            }
          ]
        }
      },
      "runId": "00005015-0000-0016-2e04-8a6dbdde4eb9"
    },
    "schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent"
  }
}
{
  "from": "<pod address>",
  "path": "/api/v1/lineage",
  "content_type": "application/json",
  "body": {
    "eventTime": "2026-10-08T14:28:04.861821+00:00",
    "eventType": "FAIL",
    "inputs": [
      {
        "facets": {},
        "inputFacets": {},
        "name": "lineage-src/orders.bad",
        "namespace": "s3://docs-src"
      }
    ],
    "job": {
      "facets": {
        "jobType": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.53.0/client/python",
          "_schemaURL": "https://openlineage.io/spec/facets/2-0-4/JobTypeJobFacet.json#/$defs/JobTypeJobFacet",
          "integration": "VAST_DATAENGINE",
          "jobType": "FUNCTION",
          "processingType": "STREAMING"
        }
      },
      "name": "lineage-demo.lineagefn",
      "namespace": "vast-dataengine://my-tenant"
    },
    "outputs": [],
    "producer": "https://example.com/lineage-demo",
    "run": {
      "facets": {
        "errorMessage": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.53.0/client/python",
          "_schemaURL": "https://openlineage.io/spec/facets/1-0-1/ErrorMessageRunFacet.json#/$defs/ErrorMessageRunFacet",
          "message": "cannot process lineage-src/orders.bad",
          "programmingLanguage": "python"
        },
        "tags": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.53.0/client/python",
          "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/TagsRunFacet.json#/$defs/TagsRunFacet",
          "tags": [
            {
              "key": "openlineage_client_version",
              "source": "OPENLINEAGE_CLIENT",
              "value": "1.53.0"
            }
          ]
        }
      },
      "runId": "00005015-0000-0016-2e04-8a6dbdde4eb9"
    },
    "schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent"
  }
}
…

Emitting FAIL doesn't change how DataEngine treats the failed event: it follows the link's retries and the dead-letter topic as usual (see dead-letter topic). With retries, each attempt would emit its own START and FAIL.

Send events through an Event Broker topic

Instead of reaching a backend over HTTP, a function can publish its lineage events to a VAST Event Broker topic with the client's Kafka transport. The traffic then stays between the compute cluster and the VAST cluster, and anything that reads OpenLineage events from Kafka can consume them, including OpenMetadata (see OpenMetadata).

Create the topic, here lineage-events, on your tenant's broker (see topics). The example used one partition.
Add the Kafka extra to lineagefn/requirements.txt:
```
openlineage-python[kafka]==1.53.0
```
Build the image as v2, and publish it as the function's next revision (see release a new version). The function code doesn't change.

vastde functions build "$REGISTRY/docs/lineagefn" --target ./lineagefn --image-tag v2 --push

Detected language: python
Validating Python version 3.12.*...
Python version 3.12.* resolved to 3.12.13
Building 127.0.0.1:5000/docs/lineagefn:v2
App Path: ./lineagefn
Handlers File: main.py
2026/10/08 14:29:25 [Started] Python Builder: 127.0.0.1:5000/docs/lineagefn:v2
2026/10/08 14:30:10 [Completed] Python Builder: 127.0.0.1:5000/docs/lineagefn:v2
Build completed: 127.0.0.1:5000/docs/lineagefn:v2

Then select the Kafka transport with OPENLINEAGE__TRANSPORT__* variables, in place of OPENLINEAGE_URL. OPENLINEAGE__TRANSPORT__CONFIG holds the Kafka producer settings as JSON, with the Event Broker's address as bootstrap.servers. Save this as lineage-demo-kafka.yaml:

name: lineage-demo
description: emits OpenLineage run events for every object
kubernetes_cluster_vrn: vast:dataengine:kubernetes-clusters:test-cluster
namespace: dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91
manifest:
  triggers:
    - name: uploads
      vrn: vast:dataengine:triggers:lineage-uploads
  function_deployments:
    - name: lineagefn
      function_vrn: vast:dataengine:functions:lineagefn
      config:
        environment_variables:
          - name: OPENLINEAGE__TRANSPORT__TYPE
            value: kafka
          - name: OPENLINEAGE__TRANSPORT__TOPIC
            value: lineage-events
          - name: OPENLINEAGE__TRANSPORT__CONFIG
            value: '{"bootstrap.servers": "<Event Broker address>:9092", "acks": "all"}'
          - name: LINEAGE_NAMESPACE
            value: vast-dataengine://my-tenant
  links:
    - source: [uploads]
      destination: [lineagefn]
      config:
        retries: 0

The pipeline doesn't pin a function revision, so deploying it picks up revision 2:

vastde pipelines update lineage-demo --config @lineage-demo-kafka.yaml && vastde pipelines deploy lineage-demo

Pipeline updated successfully
Name: lineage-demo
Description: emits OpenLineage run events for every object
Tags: []
GUID: 158b72ff-ba03-49d9-8feb-fb070173fc6a
Owner: [id: 4294967295, id-type: vid]
Created At: 2026-10-08T14:27:03Z
Updated At: 2026-10-08T14:30:38Z
VRN: vast:dataengine:pipelines:lineage-demo
Status: Ready

Pipeline deployed successfully

until s=$(vastde pipelines get lineage-demo -o json | jq -r .status); [ "$s" = Ready ] || [ "$s" = Error ]; do sleep 5; done; vastde pipelines get lineage-demo -o json | jq -c "{status, reason}"

{"status":"Ready","reason":"Pipeline is ready and active."}

aws --region us-east-1 --endpoint-url "$S3_ENDPOINT" s3 cp orders.csv "s3://$BUCKET/lineage-src/orders-2.csv" >/dev/null; sleep 15; vastde logs get lineage-demo --since 2m --scope user

2026-10-08 14:31:11.66 [lineagefn] [INFO]  [user] openlineage.client.client: OpenLineageClient will use `kafka` transport
2026-10-08 14:31:11.66 [lineagefn] [INFO]  [user] lineage job vast-dataengine://my-tenant/lineage-demo.lineagefn
2026-10-08 14:31:37.70 [lineagefn] [INFO]  [user] processing lineage-src/orders-2.csv, lineage run 00005015-0000-0018-aad6-8e1d107948bc, trace 0000501500000018aad68e1d107948bc

Read the topic with the read_topic.py script from read the dead-letter topic, cutting each line to 400 characters. The client keys each record by its job:

python read_topic.py lineage-events | cut -c1-400

--- partition=0 offset=0 timestamp=1791469897183
  key: run:vast-dataengine://my-tenant/lineage-demo.lineagefn
  value: {"eventTime": "2026-10-08T14:31:37.182769+00:00", "eventType": "START", "inputs": [{"facets": {}, "inputFacets": {}, "name": "lineage-src/orders-2.csv", "namespace": "s3://docs-src"}], "job": {"facets": {"jobType": {"_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.53.0/client/python", "_schemaURL": "https://openlineage.io/spec/facets/2-0-4/JobTypeJobFacet.json#/$defs/JobTyp
--- partition=0 offset=1 timestamp=1791469897709
  key: run:vast-dataengine://my-tenant/lineage-demo.lineagefn
  value: {"eventTime": "2026-10-08T14:31:37.709275+00:00", "eventType": "COMPLETE", "inputs": [{"facets": {}, "inputFacets": {}, "name": "lineage-src/orders-2.csv", "namespace": "s3://docs-src"}], "job": {"facets": {"jobType": {"_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.53.0/client/python", "_schemaURL": "https://openlineage.io/spec/facets/2-0-4/JobTypeJobFacet.json#/$defs/Job
2 record(s) in lineage-events

To forward the events to a backend that only accepts HTTP, run a consumer that reads the topic and posts each record (see consuming).

OpenMetadata

From DataEngine's side, OpenMetadata is one more consumer of the lineage your functions emit. OpenMetadata documents an OpenLineage connector that reads OpenLineage events from a Kafka topic (OpenMetadata: OpenLineage connector), so the Event Broker transport above is the natural fit. That connector wasn't run for this page.

Cataloguing the VAST DataBase tables themselves in OpenMetadata, for example through OpenMetadata's Trino connector against a Trino that uses the VAST connector, isn't DataEngine work: it belongs with VAST DataBase and Trino. If you name your output tables with OpenLineage's Trino convention (see name the job, the run and the datasets), a backend that also catalogues those tables through Trino can match them to your functions' lineage.

Clean up

Delete the pipeline, the trigger and the function (vastde pipelines delete lineage-demo, vastde triggers delete lineage-uploads, vastde functions delete lineagefn), and the topic if you created one.