Data lineage (OpenLineage)

Tracing answers what happened. Lineage answers a different question: what data did this produce, and what did it come from? Hopskip speaks OpenLineage, the standard consumed by Marquez, DataHub, Astronomer, and Airflow’s own lineage backend. A workflow that reads a table your Airflow DAG wrote shows up as one connected graph in tooling neither system owns.

Nothing is exported until you configure a destination.

Turn it on

Point the server at an OpenLineage receiver:

export HOPSKIP_OPENLINEAGE_ENDPOINT=http://marquez:5000
export HOPSKIP_OPENLINEAGE_API_KEY=…            # optional bearer token
hopskip-server

Events post to {endpoint}/api/v1/lineage, the OpenLineage HTTP transport’s own convention. To try it without standing up a backend, write the events to a file instead:

export HOPSKIP_OPENLINEAGE_FILE=./lineage.jsonl
hop dev
# ... run a workflow, then:
tail -f lineage.jsonl

That alone gets you a run-level graph: every workflow run becomes an OpenLineage job (the workflow type) and run (this execution), with START and COMPLETE/FAIL/ABORT events, and a parent/child edge wherever one workflow started another.

Declare the datasets

Core cannot know which data a run touched. To the engine, an activity that copies a table into Parquet is opaque bytes in and opaque bytes out. Only you know the table and the file, so you say so:

from hopskip import durable, activity, dataset_input, dataset_output

@durable
async def daily_rollup(day):
    dataset_input("orders_source", "postgres://db:5432", "shop.public.orders")
    rows = await activity("read_orders", day)

    await activity("write_parquet", rows)
    dataset_output("rollup", "s3://warehouse", f"orders/{day}.parquet")

Those two calls are what turn a run into a node with edges on both sides. The same API exists in every SDK that has it:

Language	Input	Output
Python	`dataset_input(key, ns, name)`	`dataset_output(key, ns, name)`
TypeScript	`declareDatasetInput(key, ns, name)`	`declareDatasetOutput(key, ns, name)`
Rust	`lineage::input(key, ns, name)`	`lineage::output(key, ns, name)`
Go	`hopskip.DatasetInput(key, ns, name)`	`hopskip.DatasetOutput(key, ns, name)`

Each also has a Dataset value for a dataset you declare more than once:

let orders = Dataset::new("orders", "postgres://db:5432", "shop.public.orders");
orders.read();

Declaring the same dataset twice is harmless; the emitter deduplicates.

Naming datasets

namespace and name are OpenLineage’s coordinates, not Hopskip’s. The namespace is the source system (postgres://db:5432, s3://warehouse, bigquery), and the name is the dataset within it (shop.public.orders, orders/2026-07-30.parquet). Following OpenLineage’s naming conventions is what makes your Hopskip run connect to the same table your dbt model wrote. Inventing a local convention gives every tool its own island.

key is the @searchable attribute key the declaration rides on: your label for the call site, not part of the dataset’s identity.

What an event looks like

{
  "eventType": "COMPLETE",
  "eventTime": "2026-07-30T16:00:09.000Z",
  "job": { "namespace": "orders", "name": "hopskip:orders.fulfill@2" },
  "run": {
    "runId": "5c6af70c-96fe-8c67-a701-ca503358540e",
    "facets": {
      "nominalTime": { "nominalStartTime": "…", "nominalEndTime": "…" },
      "hopskip": {
        "taskId": "client-1",
        "workflowInstance": "order-42",
        "hopskipRunId": "019a…",
        "principal": "alice@example.com",
        "terminalStatus": "completed"
      }
    }
  },
  "inputs":  [{ "namespace": "postgres://db:5432", "name": "shop.public.orders" }],
  "outputs": [{ "namespace": "s3://warehouse", "name": "orders/2026-07-30.parquet" }]
}

The job is the workflow type, because that is what repeats across runs. The per-instance name you chose identifies the run and rides on the hopskip facet. Putting the instance name in job.name would give you a million single-run nodes and no edges worth drawing.

How it works

The emitter is a read-side consumer of the event log, the same shape as the visibility projection. It reads the durable records a run already leaves behind (its start pair, its terminal outcome, and any dataset declarations) and maps them into run events. Nothing about lineage sits on the dispatch path, so an unreachable backend can only fall behind; it can never fail a workflow or slow an append.

A tick is three phases, and only the first one touches the log: read the new records, then decode and map them, then deliver. The server holds the dispatcher and log locks for the read alone. Neither the decoding cost nor the network round trip happens on a lock your appends need.

Delivery is at-least-once, by design. The OpenLineage run id is derived (a name-based UUID over the run’s durable identity), so a re-sent event is byte-identical to the first one and lands on the same lineage node. A backend outage costs re-derivation, never lineage: nothing is consumed until it lands, so when the backend recovers the whole pending backlog arrives.

Requests are bounded (5s to connect, 15s total), so a backend that accepts a connection and never answers costs a batch rather than wedging the exporter. Consecutive failed ticks back off exponentially to about five minutes and reset the moment anything is delivered, so a long outage does not mean a tick’s worth of wasted work every few seconds for its duration.

Recovery

The exporter keeps nothing durable (offsets and mapping state are process memory), so recovery is always the same act: re-read the log and re-derive. Three situations, and what each one costs you.

A crash or restart. The restarted process re-reads every partition the log still holds and re-derives everything, including runs that had already finished. Those events are byte-identical to the ones sent before the crash, so your backend collapses them onto the same lineage nodes rather than forking them. Nothing is lost, including a run that finished in the seconds before the crash.

That last part depends on your log backend being able to enumerate its own partitions. The memory and raft backends can; disk and tiered cannot, because they key their segments by hash. On a backend that cannot, the exporter falls back to asking the dispatcher, which only knows about runs this process started, so a run that finished within one tick of a crash never gets its terminal event. The server says so once at startup rather than leaving you to find out:

WARN log backend does not enumerate partitions, so lineage recovery after a
     restart is limited to runs this process started ...

A backend outage. Nothing is consumed until it lands, so an outage is a backlog rather than a gap. When the backend comes back, the whole pending set is delivered.

Multiple nodes or regions. Two nodes reading the same replicated log emit byte-identical events, because every field is derived from the durable record and nothing from the emitting process. Your backend sees duplicates of one graph, not two graphs that disagree.

Where nodes genuinely differ is visibility. If a workflow started by a run on another node lands on this one, this node can see the child’s declared parent but not the parent’s identity, so it emits the child without a parent edge rather than guessing one. That is counted and logged rather than dropped silently:

INFO openlineage: emitted events whose declared parent run is not visible to
     this node; their lineage edges are omitted rather than guessed

A standing non-zero count is the normal shape of a multi-region deployment: each node contributes the edges it can substantiate, and it signals that the graph from any single node is that node’s view rather than the whole one.

Terminal states

Hopskip outcome	OpenLineage
Completed	`COMPLETE`
Failed, retries exhausted, poison, DLQ, quarantine	`FAIL`
Superseded by a newer start	`ABORT`

ABORT is reserved for a run that was stopped, not one that broke. The exact Hopskip status (skipped_to_dlq, quarantined, …) is preserved on the hopskip facet, so nothing is lost in the narrowing.

Configuration

Variable	Default	Meaning
`HOPSKIP_OPENLINEAGE_ENDPOINT`	(none)	Base URL of an OpenLineage receiver. Unset = no export.
`HOPSKIP_OPENLINEAGE_API_KEY`	(none)	Bearer token. Unset = no `Authorization` header.
`HOPSKIP_OPENLINEAGE_FILE`	(none)	Append events to this file as newline-delimited JSON. Ignored when an endpoint is set.
`HOPSKIP_OPENLINEAGE_TICK_MS`	`5000`	How often the emitter reads new records.
`HOPSKIP_OPENLINEAGE_BATCH_SIZE`	`500`	Records read per partition per tick.

Limits worth knowing

No column-level lineage. The engine sees opaque payload bytes, not columns. A guessed column mapping would be worse than none.
Activities are not their own runs. Promoting every step to a lineage node would fill the graph with Hopskip internals rather than your pipeline. The one relationship that is a real durable fact, one workflow starting another, is emitted as a parent/child edge.
Haskell has no dataset-declaration API yet. Haskell workflows still produce run-level lineage; they just cannot declare datasets from guest code yet.
Non-ASCII dataset names from TypeScript are declined, not corrupted, because that SDK has no UTF-8 encoder behind its scalar encoding yet. Rust, Python, and Go handle them fine.