Report lineage from Airflow, dbt and Spark

The OpenLineage component, @bijection/openlineage, is not yet distributed on npm. Its token providers are installed from its public repository, bijectionhq/openlineage.

The OpenLineage component accepts standard OpenLineage events at POST /api/lineage on your deployment’s HTTP actions host. A reporter, such as Airflow, dbt or a Spark job, sends them as one of your services. Four things take part:

A source in your defineOpenLineage declaration names the service and the job and dataset namespaces it may report.
A service in your access model is the reporter’s identity. It holds a role that permits report on its source.
A credential authenticates the service: a key bijection auth credential create writes, or a machine client of your identity provider that bijection auth credential add trusts.
A token provider in the reporter’s OpenLineage HTTP transport turns the credential into a short-lived bearer token for each request.

Declare the source

Name the service that reports for each source, and every namespace it reports:

bijection/lineage.ts

export const lineage = defineOpenLineage(components.openlineage, {
  sources: {
    airflow: {
      service: "airflow",
      jobNamespaces: ["airflow://production"],
      datasetNamespaces: ["postgres://warehouse.internal:5432"],
      retention: { events: 256, maxAgeMs: 7 * DAY },
    },
  },
}, { access: internal.lineageAccess.decide });

// bijection/http.ts
installOpenLineageRoutes(http, components.openlineage);

A service reports for at most one source. Namespaces are exact: an event whose job or any dataset carries another namespace is refused, and the refusal names the namespace to add. Airflow reports its jobs under the namespace set by AIRFLOW__OPENLINEAGE__NAMESPACE, dbt under OPENLINEAGE_NAMESPACE and Spark under spark.openlineage.namespace; dataset namespaces come from the warehouse connection, such as postgres://host:5432 or snowflake://<account>. The service is declared in your access model with a ceiling that allows reporting. The component asks your access policy about these resource kinds:

Kind	Key	Operations
`lineage_source`	the source’s name	`read`, `report`, `retain`, `read_raw`
`lineage_destination`	the destination’s name	`read`, `configure`, `export_native`, `retire`
`external_dataset`	the dataset’s entity key	`read`
`external_job`	the job’s entity key	`read`

bijection/access.ts

export const access = defineAccess({
  resources: {
    lineage_source: { permissions: ["read", "report", "retain", "read_raw"] },
    lineage_destination: {
      permissions: ["read", "configure", "export_native", "retire"],
    },
    external_dataset: { permissions: ["read"] },
    external_job: { permissions: ["read"] },
    // ...your other resources
  },
  roles: {
    "lineage_source:reporter": ["lineage_source:report"],
    // Include lineage_source:* in the role that administers access.
  },
  services: {
    airflow: {
      description: "Airflow reports production lineage",
      ceiling: ["lineage_source:report"],
    },
  },
});

Answer each request with the same-named permission:

bijection/lineageAccess.ts

import { accessGrant, accessRequest } from "bijection/components";
import { internalQuery } from "./_generated/server";
import { access } from "./access";

export const decide = internalQuery({
  args: accessRequest,
  returns: accessGrant,
  handler: async (ctx, { resource, operation }) => {
    const decision = await access.check(ctx, {
      permission: `${resource.kind}:${operation}`,
      resource: { type: resource.kind, key: resource.key },
    });
    return decision.kind === "allowed" ? { scope: resource.key } : null;
  },
});

Create a credential

A credential belongs to one deployment; create one for each of development, staging and production.

Service key
Identity provider machine client

bijection auth credential create airflow --name production

This installs the credential’s public key and writes the private key to airflow.production.servicekey. Bijection keeps no copy. Store the file’s one line as a secret where the reporter runs, then delete the file. The token providers below sign a one-minute token per minute with it.

If your identity provider already issues the reporter a machine client, trust it as the service instead:

bijection auth credential add airflow \
  --issuer https://auth.example.com/ --client <subject>

The provider must be one your deployment verifies, in bijection/auth.config.ts, and its tokens must carry the audience configured there. --client is the subject the provider’s tokens name: the client ID for most providers, <client id>@clients for Auth0. The client’s scopes decide nothing; the service’s grants do.

Either way the deployment verifies the reporter as the service airflow. bijection auth credential revoke <credential> refuses its tokens from then on.

Grant reporting

Grant the service the reporter role on its source:

bijection access grant lineage_source:reporter --to service:airflow --on lineage_source/airflow

Withdrawing the grant stops reporting at once. The grant permits nothing beyond reporting events for that source.

Configure the reporter

Install the token provider where the reporter runs, then point the OpenLineage HTTP transport at your deployment’s HTTP actions host, https://<deployment>.bijection.run, with the endpoint api/lineage.

Setting	Value
URL	`https://<deployment>.bijection.run`
Endpoint	`api/lineage` (`/api/lineage/batch` takes an array of up to 64)
Auth	A token provider below; never a static key
Body	Standard OpenLineage 2-0-2 events, JSON, optionally gzip

Airflow
dbt
Spark

Install the Python token provider next to the OpenLineage provider:

pip install "bijection-openlineage @ git+https://github.com/bijectionhq/openlineage#subdirectory=clients/python"

Then set the transport, for example as an environment variable:

AIRFLOW__OPENLINEAGE__NAMESPACE=airflow://production
AIRFLOW__OPENLINEAGE__TRANSPORT='{"type": "http", "url": "https://<deployment>.bijection.run", "endpoint": "api/lineage", "auth": {"type": "bijection_openlineage.ServiceKeyTokenProvider", "credentialFile": "/run/secrets/airflow.servicekey"}}'

credential takes the key’s one line instead of a file.

Install the Python token provider with openlineage-dbt, and run dbt through dbt-ol:

pip install openlineage-dbt "bijection-openlineage @ git+https://github.com/bijectionhq/openlineage#subdirectory=clients/python"

openlineage.yml

transport:
  type: http
  url: https://<deployment>.bijection.run
  endpoint: api/lineage
  auth:
    type: bijection_openlineage.ServiceKeyTokenProvider
    credentialFile: /run/secrets/dbt.servicekey

OPENLINEAGE_NAMESPACE=dbt://production dbt-ol run

Build the Java token provider (Java 8 or later) and add its jar next to the OpenLineage Spark listener:

git clone https://github.com/bijectionhq/openlineage
mvn -f openlineage/clients/java/pom.xml package

spark-submit \
  --jars openlineage/clients/java/target/bijection-openlineage-0.1.0.jar \
  --conf spark.extraListeners=io.openlineage.spark.agent.OpenLineageSparkListener \
  --conf spark.openlineage.namespace=spark://production \
  --conf spark.openlineage.transport.type=http \
  --conf spark.openlineage.transport.url=https://<deployment>.bijection.run \
  --conf spark.openlineage.transport.endpoint=/api/lineage \
  --conf spark.openlineage.transport.auth.type=com.bijection.openlineage.ServiceKeyTokenProvider \
  --conf spark.openlineage.transport.auth.credentialFile=/etc/secrets/spark.servicekey \
  job.py

With an identity provider’s machine client, use its client-credentials token provider instead:

Client	`auth.type`	Settings
Python (Airflow, dbt)	`bijection_openlineage.ClientCredentialsTokenProvider`	`tokenEndpoint`, `clientId`, `clientSecret`, optional `audience`, `scope`
Java (Spark), 1.54 or later	`oauth2`	`tokenEndpoint`, `clientId`, `clientSecret`, optional `scope`

The Java provider sends no audience; with Auth0, set the API as the tenant’s default audience.

When a report is refused

Every refusal names the check that failed and links the section of this page that fixes it, in error.code and error.docs of the response body (in failed_events for a batch).

Status	Code	Meaning
401	`authentication_required`	The request carried no bearer token.
401	`unknown_issuer`	No service credential or configured identity provider has the token’s issuer.
401	`token_not_verified`	The token is expired, revoked, or signed by another deployment’s credential.
403	`no_source_for_identity`	The caller is not a service, or no source declares its service.
403	`namespace_not_admitted`	The event names a job or dataset namespace its source does not list; the message names it.
403	`report_not_granted`	The service holds no grant of `report` on its source.

Refusals are not retriable as sent. Capacity and availability failures (429 and 503) are; see the component’s limits in Inspection.

Get Started

Use-case atlas

Platform

Data and Operations

Guides

Client Libraries

Tools

API Reference

Report lineage from Airflow, dbt and Spark

Declare the source

Create a credential

Grant reporting

Configure the reporter

When a report is refused

​Declare the source

​Create a credential

​Grant reporting

​Configure the reporter

​When a report is refused

Declare the source

Create a credential

Grant reporting

Configure the reporter

When a report is refused