Pipelines and applications

Use runtime events when a system can report what actually ran. Add declared artifacts and static analysis to fill gaps without overriding stronger evidence.

Airflow

Connect Airflow to see observed DAGs and tasks and associate task execution with warehouse lineage. Embrasure uses Airflow's official OpenLineage provider. You do not need an Embrasure package, callback, or change to your DAG code.

Before you start

• Connect and sync each warehouse that your Airflow tasks read from or write to. Embrasure resolves datasets against those catalogs.
• Use an Embrasure account with permission to manage connections.
• Have the URL of your Airflow UI available if you want links from Embrasure back to Airflow.

1. Create the Airflow connection

Open Connections → Other, then select Add Airflow.

1. Give the connection a name such as Production Airflow.
2. Optionally enter the URL of the Airflow UI.
3. Create the connection.

Create a separate connection for each Airflow deployment. Embrasure generates a unique OpenLineage namespace and a connection-scoped token so DAGs with the same name remain distinct across deployments.

2. Install the OpenLineage provider

Add the official provider to the same Python environment or container image as Airflow:

pip install apache-airflow-providers-openlineage

Use a provider version compatible with your Airflow version. Redeploy or restart the Airflow scheduler and workers after installing it.

3. Configure Airflow

Copy the two environment variables shown by Embrasure immediately after creating the connection. The token is displayed only once.

The generated values have this structure:

export AIRFLOW__OPENLINEAGE__NAMESPACE='<generated namespace>'
export AIRFLOW__OPENLINEAGE__TRANSPORT='{"type":"http","url":"https://api.embrasure.ai","endpoint":"api/v1/airflow/<connection-id>/lineage","auth":{"type":"api_key","apiKey":"<connection token>"}}'

Set both variables for the scheduler and workers, then restart them. Store the token with your normal Airflow secrets. If it is lost or exposed, generate a new token from the connection in Embrasure and replace the transport configuration.

4. Run a DAG

Run a DAG normally. The provider emits OpenLineage events automatically. Embrasure creates the DAG and task records when those events arrive and adds table lineage when an event includes datasets that match a connected catalog.

5. Verify the connection

Return to Connections → Other and open the Airflow connection.

• The status changes from Waiting for Airflow to Events received.
• Observed DAGs lists the DAGs and tasks Embrasure has received.
• The lineage view attributes matching table relationships to Airflow and shows the DAG and task hierarchy in the edge details.

If no events arrive, confirm that the provider is installed in the running Airflow environment, both variables are available to the scheduler and workers, and the processes were restarted. Airflow task logs will also show failed OpenLineage deliveries.

How the connection is scoped

• The connection records new OpenLineage events; it does not import historical DAGs or runs from the Airflow API.
• One task may read from several connected warehouses. A single event may write to one connected destination warehouse.
• Dataset names must resolve to tables in connected and synced catalogs. Events with unknown or ambiguous datasets are rejected instead of being attached to the wrong table.
• Keep the generated namespace stable for the lifetime of the Airflow deployment.

Spark

Dataproc can report lineage without the listener through Google Data Lineage; configure that on the cloud and warehouses page. To send events directly from any Spark runtime, use the OpenLineage listener. Its transport configuration is the same on every platform:

spark.extraListeners=io.openlineage.spark.agent.OpenLineageSparkListener
spark.openlineage.transport.type=http
spark.openlineage.transport.url=https://api.embrasure.ai
spark.openlineage.transport.endpoint=/api/v1/lineage?workspace_id=WORKSPACE_ID&connector_id=CONNECTOR_ID
spark.openlineage.transport.auth.type=api_key
spark.openlineage.transport.auth.apiKey=EMBRASURE_API_TOKEN
spark.openlineage.namespace=spark-production

What differs by platform is how the listener reaches the classpath:

• Dataproc and Glue 5.0 bundle the listener in the runtime. Pass the properties above as job or cluster Spark configuration.
• EMR, Glue 3.0 and 4.0, Synapse, Kubernetes, and self-managed clusters also need the openlineage-spark jar on the classpath, for example through --extra-jars, --packages, or a bootstrap action.
• Platform-native lineage toggles publish to the cloud provider's own catalog. Embrasure reads Dataproc's links through the BigQuery connector's Google Data Lineage sync; for other platforms, the listener is the path to Embrasure.

Flink and other runtime producers can send standard OpenLineage events to the same lineage receiver.

dbt

Upload target/manifest.json after dbt compile or dbt build:

: "${EMBRASURE_LINEAGE_ACQUISITION_ID:?set a unique acquisition ID}"

jq --arg workspace "$WORKSPACE_ID" \
   --arg connector "$CONNECTOR_ID" \
   --slurpfile manifest target/manifest.json \
   '{workspace_id:$workspace, connector_id:$connector, manifest:$manifest[0], completeness:"complete"}' \
  > /tmp/embrasure-dbt-manifest.json

curl --fail-with-body --request POST \
  --url https://api.embrasure.ai/v1/lineage/dbt/manifest \
  --header "Authorization: Bearer $EMBRASURE_API_TOKEN" \
  --header "Content-Type: application/json" \
  --header "Idempotency-Key: dbt-$BUILD_ID" \
  --data-binary @/tmp/embrasure-dbt-manifest.json

Use complete only for the exact project and connector scope. Use a new idempotency key for each acquisition and reuse it only to retry identical bytes.

SQLMesh

Run the project export as the baseline: it captures the complete declared project for one environment, including models that did not run, and maps each virtual model name to its physical snapshot so other evidence resolves to the right asset. Then add runtime events for executed-run evidence, which carries more authority than declarations when both exist for the same edge.

Export and upload

Export the project after SQLMesh applies the target environment, then submit the rendered project to POST /v1/lineage/sqlmesh/project. Run the exporter with the same Python interpreter that runs SQLMesh, after sqlmesh plan --auto-apply or your normal deployment step. It needs only the standard library and your existing SQLMesh install.

Download the version pinned to the embrasure.sqlmesh-project@v1 export schema into your SQLMesh project:

curl --fail --location \
  --output embrasure_sqlmesh_export.py \
  https://raw.githubusercontent.com/EmbrasureAI/embrasure/67665401b7bb08d49d4912f9d65d8204fe847c39/scripts/sqlmesh/embrasure_sqlmesh_export.py

mkdir -p target
python embrasure_sqlmesh_export.py \
  --project . \
  --environment prod \
  --output target/embrasure-sqlmesh-project.json

The exporter resolves environment-specific physical snapshot names from SQLMesh state. Embrasure keeps each virtual model name as the stable asset identity and records its physical snapshot as an alternate identifier. If state is unavailable, the exporter warns on stderr. Virtual table lineage still imports, but physical-name merging is unavailable.

Wrap the export with its workspace, connector, and project identities, then upload it:

jq --arg workspace "$WORKSPACE_ID" \
   --arg connector "$CONNECTOR_ID" \
   --slurpfile project target/embrasure-sqlmesh-project.json \
   '{workspace_id:$workspace, connector_id:$connector, project_id:"analytics-sqlmesh", completeness:"complete", project:$project[0]}' \
  > /tmp/embrasure-sqlmesh-request.json

curl --fail-with-body --request POST \
  --url https://api.embrasure.ai/v1/lineage/sqlmesh/project \
  --header "Authorization: Bearer $EMBRASURE_API_TOKEN" \
  --header "Content-Type: application/json" \
  --header "Idempotency-Key: sqlmesh-$EMBRASURE_LINEAGE_ACQUISITION_ID" \
  --data-binary @/tmp/embrasure-sqlmesh-request.json

Use complete only for an entire project in one environment. The route accepts partial or complete, honors Idempotency-Key, and returns the standard lineage delivery receipt. Set the acquisition ID to a unique per-run value, such as $GITHUB_RUN_ID-$GITHUB_RUN_ATTEMPT in GitHub Actions. Every acquisition needs a new key, including a return to previously uploaded content. Reuse a key only to retry the same request.

Rendered SQL supplies column mappings only when every read matches a declared dependency. Python models and models without rendered SQL keep table-level dependencies without guessed column lineage. See the SQLMesh project API reference for the request contract.

Runtime events (recommended addition)

The community-maintained sqlmesh-openlineage package emits a run event per executed model evaluation, adding executed-run evidence on top of the export. It is not maintained by Embrasure, so pin and test its version with your SQLMesh release; it requires Python-based SQLMesh configuration. If you cannot install third-party instrumentation, project export remains the fully supported path.

Install the package, then enable it in config.py before creating the SQLMesh Config:

pip install sqlmesh-openlineage

import sqlmesh_openlineage

sqlmesh_openlineage.install()

Use the bare API origin because OpenLineage clients append /api/v1/lineage. Authenticate with a workspace-scoped personal access token that has write scope, and keep the namespace stable for one deployment:

export OPENLINEAGE_URL="https://api.embrasure.ai"
export OPENLINEAGE_NAMESPACE="analytics-sqlmesh-prod"
export OPENLINEAGE_API_KEY="$EMBRASURE_WORKSPACE_SCOPED_TOKEN"

Run sqlmesh run normally. See the lineage receiver for authentication and dataset resolution.

Fivetran

Connect Fivetran in the app: open Settings → Integrations → Fivetran, enter a Fivetran API key and secret, and validate them. Then map each Fivetran connection to its Embrasure source and destination connectors and enable it. Embrasure currently syncs Postgres-to-BigQuery replication lineage, recording connector assets, sync runs, table mappings, and enabled column mappings.

DataHub and custom applications

Import DataHub exports at POST /v1/lineage/datahub. For applications and APIs, send standard OpenLineage datasets, jobs, runs, and optional columnLineage facets to the lineage receiver.

Use stable, case-sensitive namespaces and names. Both endpoints must resolve to cataloged objects before an edge appears in the served graph.