Skip to content

Leveraging Open Lineage

Most processing engines - Apache Spark, Apache Airflow, dbt, ETL tools and countless custom jobs - either produce no lineage information at all, or produce it in their own proprietary format. Open Lineage is a sister open source project to Egeria in the LF AI and Data Foundation that gives these engines a common, vendor-neutral way to report what they did: which data they read, what they wrote, and how the two are connected.

Egeria is able to receive, store, route and act on Open Lineage events. This means it can:

  • Harvest information about the data stores and processes that are emitting Open Lineage events, adding them to open metadata as they are discovered.
  • Store the Open Lineage events in an organized way, either in its own file-based log store or in an external Open Lineage-compliant server such as Marquez, so they are available for later analysis.
  • Route Open Lineage events to other services - both events received from external processing engines, and events that Egeria generates itself from the governance actions it runs.

This makes Open Lineage one of the main ways that operational (dynamic, "what actually ran") lineage gets into Egeria's lineage graph, alongside the design (static) lineage captured by cataloguing. See Lineage Management for the complete picture of how captured lineage - from Open Lineage and other sources - is stitched together and preserved.

The Open Lineage standard

When a processing engine such as Apache Spark runs a process, it produces a series of RunEvents describing the activity of that process. The Open Lineage standard defines the format of these events and a single, simple REST API operation, {{urlroot}}/api/v1/lineage, that receives them.

The Open Lineage standard defines the payload for RunEvents as well as a standard URL for a service that acts as a collection point for them.

A RunEvent has eight parts:

  • eventType - the type of activity being described.
  • eventTime - the time of the event.
  • run - the description of the process instance.
  • job - the description of the process.
  • inputs - the data sources used as inputs by the process instance.
  • outputs - the data sources that hold the output of the process instance.
  • producer - the name/location of the processing engine producing the events.
  • schemaURL - the location of the JSON schema that describes the structure of the RunEvent.

The structure of a RunEvent.

Events also carry additionalProperties called facets - extensions that add detail such as documentation links, schema information, SQL text or data quality metrics. Any organization or processing engine can define its own custom facets alongside the standard ones.

See The Open Lineage Standard for the full set of standard facets and how they are structured.

Egeria's Open Lineage support

Egeria offers two ways to capture Open Lineage events from processing engines, depending on how the engine publishes them.

Many processing engines publish through the Open Lineage project's own proxy backend - a lightweight side-car that receives events over the API and republishes them to a Kafka topic. Egeria's Open Lineage Event Receiver integration connector listens on that topic:

Receiving events via the Kafka topic populated by the proxy backend.

Alternatively, Egeria's integration daemon can implement the Open Lineage API directly, so a local processing engine can send its events straight to Egeria without needing a proxy backend or a Kafka topic in between:

Receiving events via the Open Lineage API directly into the integration daemon.

The Open Lineage connectors

However they arrive, events are handed to the integration daemon's context manager, which distributes them to whichever integration connectors have registered as listeners. Five connectors are supplied by Egeria, falling into two groups: those that acquire or create Open Lineage events, and those that process or distribute them.

Connector Name Purpose
Open Lineage Event Receiver Receives Open Lineage events from a Kafka topic and passes them, via the integration daemon, to any other connectors in the same daemon that have registered as Open Lineage listeners.
Governance Action Open Lineage Listens for engine actions executing in the open metadata ecosystem and generates the equivalent Open Lineage events for them - so Egeria's own governance processing shows up in the same lineage picture as external processing engines.
API-based Open Lineage Log Store Registers as a listener and forwards every event it receives on to a remote server that implements the Open Lineage API, such as Marquez.
File-based Open Lineage Log Store Registers as a listener and writes every event it receives to its own file, in JSON format, in a nominated folder - organized by the namespace and job name in the event.
Open Lineage Cataloguer Registers as a listener and catalogues what the events describe: jobs become Processes, datasets become data assets (matched to assets already catalogued by other connectors where possible), and inputs and outputs are linked to the process with lineage relationships. It maintains the RunMetrics classification on each process and, optionally, the DataScope classification on the data assets, captures data quality results as survey reports and catalogues each run as a TransientEmbeddedProcess. See Cataloguing the events below.

The diagram below shows all five working together in a single integration daemon:

The pre-built integration connectors supplied by Egeria.

  1. A third-party processing engine sends Open Lineage events either directly to Egeria's Open Lineage API endpoint, or via a proxy backend to a Kafka topic.
  2. The Open Lineage Event Receiver picks up events from the Kafka topic and hands them to the integration daemon's context manager.
  3. The Governance Action Open Lineage connector registers a listener for engine actions running in the open metadata ecosystem, generates Open Lineage events to represent that processing, and hands them to the context manager too - so both sources of events flow through the same pipeline.
  4. Any connector wanting to receive events - the two log store connectors and the cataloguer - registers a listener with the context manager, and from then on receives every event that arrives.
  5. The API-based and file-based Open Lineage Log Store connectors write each event to their configured destination (a remote Open Lineage API server, or a local file), while the Open Lineage Cataloguer uses each event to make sure the job and datasets it describes are represented in open metadata and linked by lineage.

Because any combination of these connectors can be configured in the same integration daemon, you can, for example, catalog processes and archive the raw events to Marquez and forward Egeria's own governance action activity into the same picture, all at once.

Cataloguing the events

The Open Lineage Cataloguer maps each run event into open metadata as it arrives:

  • The job becomes a DeployedSoftwareComponent process named DeployedSoftwareComponent::{namespace}::{name}, with its description, SQL, source code location, job type, tags and ownership taken from the job facets. Parent and root jobs from the parent run facet become processes that own it through ProcessHierarchy relationships, and job dependencies become ControlFlow relationships.
  • Each input and output dataset becomes a data asset whose type (TabularDataSet, DataFile, DataFolder, Topic or DataSet) is chosen from the dataset's namespace and facets. Before creating one, the cataloguer looks for an existing asset whose resourceName and namespacePath match the Open Lineage name and namespace - the identity that other connectors also record - and reuses it when there is a single, type-compatible match. Anything else that matches is linked to the new asset with a PeerDuplicateLink for the duplicate management process to resolve. The schema facet creates the asset's tabular schema; a rename updates the asset's names in place and a drop archives or deletes it according to the connector's delete method.
  • Lineage is recorded as DataFlow relationships from the inputs to the process and from the process to the outputs, with LineageMapping relationships between columns from the column lineage facets.
  • Run metrics - run count, failures, first and last run times, last run duration and status, rows and bytes read and written - are maintained in the RunMetrics classification of the process.
  • Optionally, the cataloguer also catalogues each run as a TransientEmbeddedProcess owned by the job's process, captures the statistics and data quality facets as a survey report per event with one annotation per assertion, and maintains the DataScope classification of each output asset from the times and statistics of the writes.

Everything the cataloguer records is structural or is a "latest value": it is designed to keep the catalog current as events stream in, while leaving anything that needs the history of many runs to the analysis services described below.

Storing events: the Open Lineage Log Store

The Open Lineage log store is a destination where events can be written so they can be queried later - both by people investigating an issue, and by governance processes validating that the operational environment is behaving as expected (see Governing expectations). The implementation is pluggable.

Using the File-based Open Lineage Log Store, the log store is simply a directory (folder) in the filesystem, with one file per event:

An example deployment of Egeria capturing and processing Open Lineage events into a file-based log store.

Using the API-based Open Lineage Log Store, the same events are instead sent to a server implementing the Open Lineage API - such as Marquez, which also provides its own API for querying the events it has captured:

The same deployment, but using Marquez as the Open Lineage log store instead of the file system.

Both can be run side by side if you want a local archive as well as a queryable service.

Analysing the log store: the Lovelace services

A stream of run events is a series of observations. Some of the most useful facts about a pipeline - how often a job really runs, how long it typically takes, whether a table is rebuilt or accumulates, what proportion of a dataset's quality checks pass - only emerge from that series. Deriving them as each event arrives would mean rewriting the process or asset on every run; deriving them from the log store on a schedule is cheaper and gives better answers.

Three Lovelace services do this analysis. They are governance action services orchestrated by the Babbage Analytical Engine, and each can be enabled independently so that a deployment runs only the analyses it wants. Each reads the file-based log store (supplied as its openLineageLogStore action target, or through the logStoreDirectory request parameter) for the last analysisWindowDays days - 30 by default - and updates only the processes and assets that it can identify uniquely from the namespace and name in the events, using the same rules as the cataloguer.

Service Reads Records
Profile OpenLineage Runs The START, COMPLETE, FAIL and ABORT events of each job's runs. The run profile in the additionalProperties of the process's RunMetrics classification: runs per day, failure rate, the mean, median and range of the interval between runs, how regular that interval is and an inferred schedule (such as HOURLY or DAILY), typical and worst-case duration, and the mean data volume per run.
Refine OpenLineage Data Scope The writes (with lifecycle, subset and statistics facets) and reads of each dataset. The DataScope classification of the asset: whether the store is REPLACED, PARTITIONED or APPENDED by its writers, the data collection window (restarted at the latest rebuild, or extended back to the earliest known write), and the rates of writes and reads.
Summarise OpenLineage Data Quality The data quality assertions of each dataset and the tests of each job. A survey report per dataset (and per job) holding a QualityAnnotation per quality dimension whose score is the pass rate over the window, plus an overall pass rate, so that the current quality of a dataset can be read without scanning the run history.

The services are supplied in the Open Lineage content pack, so a deployment that loads the pack for the connectors has them available to Babbage as well. Running the cataloguer without the analysis services still gives complete lineage and current values; running the analysis services without the cataloguer still works for processes and assets that are already catalogued, but nothing new is created.

From Open Lineage events to a connected lineage graph

Capturing Open Lineage events is only the first stage. The jobs and data sources named in the events still need to be linked to each other - and to the rest of the catalog - to produce a single connected lineage graph, a process Egeria calls stewardship (deduplication and stitching). From there, lineage can be viewed directly through open metadata queries, or exported to a Lineage Warehouse for large-scale, long-term analysis.

See Lineage Management for the full architecture, including how design and operational lineage combine, how stitching works, and how the resulting graphs are preserved and used.

Raise an issue or comment below