How-to guides

{
  "snapshotMode": "BATCH_SNAPSHOT",
  "allowedJobNamespaces": ["github://my-org/ds-quality"],
  "systemPolicies": [
    { "match": "redshift://*", "creation": "NEVER", "entityCreation": false, "cleanupMode": "NONE" },
    { "match": "s3://*", "creation": "IF_MISSING", "technology": "Amazon S3", "entityCreation": true, "cleanupMode": "DAYS", "cleanupDays": 30 }
  ],
  "quality": { "enabled": false },
  "cleanup": { "enabled": true, "maxRemovalsPerRun": 50 },
  "outputOptions": { "dryRun": false, "targetConnectionName": "blindata-prod" }
}

---
name: openlineage-lineage
description: Writes and updates the OpenLineage JobEvent files in lineage/ that describe how this repository moves data, at column level wherever the code allows it. Use when asked to document, generate, update or check data lineage, or when a change adds, removes or modifies a step that reads or writes a dataset.
---

# OpenLineage lineage of this repository

You describe how the code of this repository moves data, as OpenLineage JobEvents
that Blindata publishes as catalog and lineage. Your goal is the **most precise
lineage the code proves**: column to column wherever you can trace it, table to
table only where you cannot. A description of the overall process is not enough.

## What a step is

A step is a unit of code that writes a dataset: a script, a function, a SQL
statement or file, a notebook, a dbt model, a Glue or Lambda handler. Each step
is one JobEvent in one file: `lineage/<job name>.json`.

- Job namespace: `github://my-org/ds-quality`, the same in every file.
- Job name: `<module>.<function>` or `<pipeline>.<step>`. Keep it stable:
  renaming it removes the old job in Blindata and creates a new one.
- A step that only reads (a report, an API response, a log) is not a step.
  Search indexes, model endpoints and caches are not datasets.

## Datasets

Name every dataset as OpenLineage does:

| Store | Namespace | Name |
| --- | --- | --- |
| Redshift | `redshift://<cluster>.<region>:<port>` | `<database>.<schema>.<table>` |
| PostgreSQL | `postgres://<host>:<port>` | `<database>.<schema>.<table>` |
| S3 | `s3://<bucket>` | the key of the dataset folder, without partitions and file names |
| Glue Data Catalog | `arn:aws:glue:<region>:<account>` | `table/<database>/<table>` |
| Athena | `awsathena://athena.<region>.amazonaws.com` | `<catalog>.<database>.<table>` |

Resolve names to their real values: follow constants, configuration files,
environment variables and infrastructure code (Terraform, CDK, CloudFormation,
serverless.yml). Use the production values. If a name cannot be resolved, do not
guess it: leave the dataset out and report it.

## How to reach column lineage

For each step, use the first approach that applies.

1. **The step writes with SQL** and the statement is known, or can be rebuilt
   because its template parameters are known: put the statement, as executed,
   in the `sql` job facet with its `dialect`, and list `inputs` and `outputs`.
   Do not add field mappings: Blindata parses the SQL and derives the column
   lineage. Expand templates (Jinja, f-strings, format) into the final SQL.
   A step that runs several writing statements becomes one job per statement.

2. **The step transforms data in code** (pandas, PySpark DataFrames, dicts,
   ORM objects): trace every output column back to its input columns and write
   the `lineage` job facet, one entry per output dataset, with `fields`:
   - Follow the data through the whole step: reads, renames, selects, computed
     columns, joins, aggregations, filters, writes.
   - When the code reads with a SQL query and then transforms the result, the
     query does not write the output: do not use the `sql` facet. Trace each
     output column through the code and the query back to the source table
     columns, and write the result in `fields`.
   - Set `transformations` on each input field:
     - `DIRECT` / `IDENTITY`: copied or renamed.
     - `DIRECT` / `TRANSFORMATION`: computed from the input; add a short
       `description` of the formula.
     - `DIRECT` / `AGGREGATION`: sum, count, average and similar.
   - Columns that influence the result without flowing into it (join keys,
     filters, group by, sort, window, conditions) go in the `inputs` of the
     entry, with their `field` and an `INDIRECT` transformation of subtype
     `JOIN`, `FILTER`, `GROUP_BY`, `SORT`, `WINDOW` or `CONDITIONAL`.
   - A pass-through (`select *`, a DataFrame written as read) is expanded
     column by column when the input schema is known from DDL, models, schema
     files or the catalog.

3. **Only when no column can be traced** (opaque libraries, schemas built at
   runtime), declare the table flow alone: the entry with its `inputs` and no
   `fields`. Report why.

Every mapping must point to code you can cite. A guessed mapping is worse than
none: when unsure, leave the field out and report it.

## Schemas and descriptions

- Each output carries the `schema` dataset facet with the columns the code
  writes, their types and order. Nested structures are nested `fields`.
- The job carries a `documentation` facet: what the step does and why, in one
  or two sentences for a business reader.
- Do not describe inputs that this repository does not write: the catalog
  already has them.

## Format

- One JobEvent per file, with `producer` and `schemaURL`
  (`https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/JobEvent`).
- Every facet has `_producer` and `_schemaURL`. Use the facet versions of
  OpenLineage 1.55.0.
- Add a `jobType` facet: `processingType` `BATCH` or `STREAMING`, the
  `integration` (for example `ECS`, `LAMBDA`, `GLUE`), `jobType` `TASK`.
- Do not write `eventTime` or `sourceCodeLocation`: the pipeline adds them.
- Never write secrets, credentials, connection strings with passwords, or
  personal data.

## When the code changes

- Update the file of each step the change touches, in the same pull request.
- A step deleted from the code: delete its file.
- Keep job names as they are unless the step really became another one.

## Your report

End with a table: step, file, lineage level reached (`column`, `partial
column`, `table`), and what you could not resolve or left out, with the file
and line that caused it. These are the points the reviewer checks first.

name: Lineage
on:
  pull_request:
    paths: ["lineage/**"]
  push:
    branches: [main]
    paths: ["lineage/**"]

jobs:
  publish:
    # A runner that reaches the agent, for example a self-hosted runner in its network.
    runs-on: self-hosted
    steps:
      - uses: actions/checkout@v4

      - name: Build the batch
        run: |
          jq -s \
            --arg now "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
            --arg repo "${{ github.server_url }}/${{ github.repository }}" \
            --arg sha "${{ github.sha }}" \
            'map(
               .eventTime = $now
               | .job.facets.sourceCodeLocation = ({
                   "_producer": "https://github.com/my-org/lineage-pipeline/v1",
                   "_schemaURL": "https://openlineage.io/spec/facets/1-1-0/SourceCodeLocationJobFacet.json#/$defs/SourceCodeLocationJobFacet",
                   "type": "git", "url": $repo, "repoUrl": $repo, "version": $sha
                 } + (.job.facets.sourceCodeLocation // {}))
             )' lineage/*.json > batch.json
          echo "Events in the batch: $(jq length batch.json)"          

      - name: Send the batch
        env:
          TENANT: ${{ secrets.BLINDATA_TENANT }}
          USER_NAME: ${{ secrets.BLINDATA_USER }}
          API_KEY: ${{ secrets.BLINDATA_API_KEY }}
          AGENT: ${{ vars.OPENLINEAGE_AGENT_URL }}
          UUID: ${{ github.event_name == 'pull_request' && vars.OPENLINEAGE_DRY_RUN_UUID || vars.OPENLINEAGE_LIVE_UUID }}
        run: |
          status=$(gzip -c batch.json | curl -sS -o response.json -w '%{http_code}' -X POST \
            "$AGENT/agent-api/v1/openlineage/$UUID/api/v1/lineage/batch" \
            -H "Content-Type: application/json" -H "Content-Encoding: gzip" \
            -H "X-BD-Tenant: $TENANT" -H "X-BD-User: $USER_NAME" -H "X-BD-ApiKey: $API_KEY" \
            --data-binary @-)
          echo "HTTP $status"
          if [ "$status" != "204" ]; then cat response.json; exit 1; fi          

{
  "snapshotMode": "ADDITIVE",
  "allowedJobNamespaces": ["airflow-prod"],
  "cleanup": { "enabled": true, "staleJobDays": 30 },
  "outputOptions": { "dryRun": false, "targetConnectionName": "blindata-prod" }
}

{
  "snapshotMode": "ADDITIVE",
  "allowedJobNamespaces": ["quality-checks"],
  "quality": { "enabled": true, "metricsEnabled": true },
  "outputOptions": { "dryRun": false, "targetConnectionName": "blindata-prod" }
}

{
  "eventType": "COMPLETE",
  "eventTime": "2026-11-04T06:00:00Z",
  "producer": "https://github.com/my-org/quality-checks/v1",
  "schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent",
  "run": { "runId": "0193a8e2-6c1f-7d2e-9b55-3f1d2a4c8e10" },
  "job": { "namespace": "quality-checks", "name": "daily_kpi_checks" },
  "inputs": [ {
    "namespace": "redshift://my-dwh.eu-west-1:5439",
    "name": "dwh.analytics.daily_kpi",
    "inputFacets": {
      "dataQualityAssertions": {
        "_producer": "https://github.com/my-org/quality-checks/v1",
        "_schemaURL": "https://openlineage.io/spec/facets/1-1-0/DataQualityAssertionsDatasetFacet.json#/$defs/DataQualityAssertionsDatasetFacet",
        "assertions": [
          { "assertion": "not_null", "success": true, "column": "plant_id" },
          { "assertion": "avg_value_in_range", "success": false, "column": "avg_value" }
        ]
      },
      "dataQualityMetrics": {
        "_producer": "https://github.com/my-org/quality-checks/v1",
        "_schemaURL": "https://openlineage.io/spec/facets/1-0-3/DataQualityMetricsInputDatasetFacet.json#/$defs/DataQualityMetricsInputDatasetFacet",
        "rowCount": 18250,
        "columnMetrics": { "plant_id": { "nullCount": 0, "distinctCount": 50 } }
      }
    }
  } ],
  "outputs": []
}