PyPIDataupdated 08 Aug 2026

openlineage-python

Item: openlineage-python
Rating: 4.3
Author: Keyoor

openlineage-python is the reference Python client for emitting OpenLineage events about data jobs, runs, inputs, outputs, schemas, and other metadata. It builds objects that follow the OpenLineage specification and sends them through configurable HTTP, Kafka, file, console, cloud-lineage, or composite transports. It records lineage facts; it does not store, query, visualize, or automatically infer those facts from arbitrary Python code.

openlineage-python · v1.52.0 · 6.2M/wk · ★ 2,586 · pushed 08 Aug 2026 · OpenLineage/OpenLineage · docs · reviewed by Keyoor

Verdict

Use openlineage-python when emitting a shared lineage protocol is the architectural goal and you already know where the events will be collected. Do not mistake the client for an automatic lineage extractor or a metadata platform.

API stability3/5The OpenLineage event concepts and transport model are stable, but the Python surface is actively migrating: openlineage.client.run emits a deprecation warning in 1.52.0 in favor of event_v2, direct URL construction is deprecated, and generated facet classes track evolving specification versions. Consumers should isolate event construction and pin versions during migrations.

Docs4/5The official site documents installation, optional dependencies, configuration precedence, transports, client use, naming, facets, and integration matrices, and the specification itself is public OpenAPI. The Python README inside the monorepo is only a pointer, so practical answers require moving between client docs, transport pages, generated API classes, and the broader specification.

Maintenance5/5PyPI 1.52.0 was published on 2026-07-23. The OpenLineage monorepo had 2,586 stars, 347 open issues and PRs, and a push on 2026-08-08, with development backed by the LF AI & Data community. The shared repository is active, though its issue count includes the specification, integrations, Java client, and other work beyond this Python package.

Ecosystem5/5The project documents integrations with Airflow, Spark, dbt, Flink, and more, while transports cover HTTP, Kafka, files, composite delivery, Google Cloud lineage, Amazon DataZone, Datadog, and MSK IAM. The protocol has a reference backend in Marquez and adoption by commercial catalogs, which is the main reason to choose it over a vendor-specific emitter.

Use it if

You operate a Python scheduler, ETL framework, or data service that must emit vendor-neutral lineage events
You already have an OpenLineage-compatible collector such as Marquez or a managed data catalog endpoint
You need standard run, job, input, output, and facet payloads rather than a proprietary metadata client
You want one event model with selectable HTTP, Kafka, file, console, or cloud transports

Skip it if

You expect automatic lineage from arbitrary code: the client sends the events you construct, while extraction and framework-specific hooks live in separate integrations
You do not have a collector or plan for the emitted JSON: this package has no lineage database, graph, search UI, or retention layer
You need strong delivery guarantees from a synchronous library call alone: transport retries and timeouts exist, but your process still owns shutdown, buffering, idempotency, and failure policy
Your organization is committed to one catalog's richer proprietary model and does not need an interoperability boundary; mapping every native concept into OpenLineage facets can become duplicate work
You want a tiny dependency for one webhook: the base client installs attrs, dateutil, YAML, requests, HTTPX, and packaging, while Kafka and cloud transports add larger optional stacks

Setup reality

pip install openlineage-python installs the event model, HTTP clients, configuration loader, and base transports; 1.52.0 requires Python 3.10 or newer. Before writing code, decide where events go. The client can load YAML through OPENLINEAGE_CONFIG, accept a config dictionary, or use legacy environment variables, but constructor forms that pass url directly and OpenLineageClient.from_environment() are deprecated in the published source. HTTP needs a valid base URL, normally posts to api/v1/lineage, defaults to a five-second timeout, verifies TLS, and supports API-key or JWT auth. Kafka requires the kafka extra; fsspec-backed file targets, Google Cloud lineage, DataZone, and MSK IAM each have their own extras and credentials. Event design is the harder setup: namespaces must be stable across producers, dataset names must identify the same physical data consistently, runId must be a valid UUID, and eventTime must be an ISO date-time rather than a date. A normal run needs one START and one terminal COMPLETE, FAIL, or ABORT event using the same run ID. Facets have versioned generated classes and schema URLs, so use the current event_v2 and generated modules instead of hand-building dictionaries or importing the deprecated openlineage.client.run module. Delivery errors need an explicit policy: decide whether lineage failure should fail the data job, configure retries and timeouts, and call client.close() so asynchronous or buffered transports can flush. Test first with console or file transport and inspect payloads before pointing production at a collector; bad names and missing terminal events are valid JSON but poor lineage.

Patterns

Create an HTTP client with API-key authenticationconfigure-http-transport

import os
from openlineage.client import OpenLineageClient

client = OpenLineageClient(config={
    'transport': {
        'type': 'http',
        'url': 'https://lineage.example.com',
        'timeout': 10,
        'auth': {'type': 'api_key', 'apiKey': os.environ['OPENLINEAGE_API_KEY']},
    }
})

Pass current configuration through config; the older OpenLineageClient(url=...) constructor form is deprecated.

Emit a START event with the v2 modelemit-run-start

from datetime import datetime, timezone
from uuid import uuid4
from openlineage.client.event_v2 import Job, Run, RunEvent, RunState, set_producer

set_producer('https://github.com/acme/orders')
run_id = str(uuid4())
client.emit(RunEvent(
    eventTime=datetime.now(timezone.utc).isoformat(),
    eventType=RunState.START,
    run=Run(run_id),
    job=Job('airflow-prod', 'orders.daily'),
))

Keep run_id for the terminal event; generating a new ID on completion creates two unrelated runs.

Attach input and output datasetsdeclare-input-output

from openlineage.client.event_v2 import InputDataset, OutputDataset

event = RunEvent(
    eventTime=now, eventType=RunState.START,
    run=Run(run_id), job=Job('airflow-prod', 'orders.daily'),
    inputs=[InputDataset('postgres://warehouse', 'raw.orders')],
    outputs=[OutputDataset('s3://analytics', 'curated/orders/')],
)
client.emit(event)

Namespace and name form dataset identity; choose a documented naming convention and keep it stable across every producer.

Close a run with the same identityemit-run-complete

client.emit(RunEvent(
    eventTime=datetime.now(timezone.utc).isoformat(),
    eventType=RunState.COMPLETE,
    run=Run(run_id),
    job=Job('airflow-prod', 'orders.daily'),
    inputs=inputs, outputs=outputs,
))

Each START should have one terminal COMPLETE, FAIL, or ABORT event; use the same job and run ID.

Emit a failed terminal stateemit-run-failure

try:
    run_pipeline()
except Exception:
    client.emit(RunEvent(
        eventTime=datetime.now(timezone.utc).isoformat(),
        eventType=RunState.FAIL,
        run=Run(run_id), job=job, inputs=inputs, outputs=outputs,
    ))
    raise

Emitting FAIL records lineage state but should not swallow the original job exception.

Describe an input dataset schemaattach-schema-facet

from openlineage.client.generated.schema_dataset import (
    SchemaDatasetFacet, SchemaDatasetFacetFields,
)

schema = SchemaDatasetFacet(fields=[
    SchemaDatasetFacetFields('order_id', 'BIGINT', ordinal_position=1),
    SchemaDatasetFacetFields('created_at', 'TIMESTAMP', ordinal_position=2),
])
input_ds = InputDataset('postgres://warehouse', 'raw.orders', facets={'schema': schema})

Facet classes are generated from versioned specifications; import them from generated modules rather than inventing a payload shape.

Print events while developingconfigure-console-transport

from openlineage.client import OpenLineageClient

client = OpenLineageClient(config={
    'transport': {'type': 'console'}
})
client.emit(event)

Console transport writes through Python logging at INFO level; configure a handler and level or the JSON may not be visible.

Write newline-delimited events to a local filewrite-events-to-file

client = OpenLineageClient(config={
    'transport': {
        'type': 'file',
        'log_file_path': 'var/openlineage/events.jsonl',
        'append': True,
    }
})

Remote object stores may not support real append; file transport needs the fsspec extra for non-local URL schemes.

Fan events out with a composite transportsend-to-two-transports

client = OpenLineageClient(config={
    'transport': {
        'type': 'composite',
        'continue_on_failure': True,
        'continue_on_success': True,
        'transports': [
            {'type': 'http', 'url': 'https://lineage.example.com'},
            {'type': 'file', 'log_file_path': 'events.jsonl', 'append': True},
        ],
    }
})

The two continuation flags control fail-fast, first-success, and fan-out behavior; choose them as an explicit delivery policy.

Load the client configuration path from the environmentconfigure-from-environment

# shell
export OPENLINEAGE_CONFIG=/etc/openlineage/client.yml

# Python
from openlineage.client import OpenLineageClient
client = OpenLineageClient()

OpenLineageClient.from_environment() is deprecated; the no-argument constructor already resolves supported environment configuration.

Inspect the exact event JSON before sendingserialize-event-json

from openlineage.client.serde import Serde

payload = Serde.to_json(event)
print(payload)

Inspecting serialized output catches naming and facet mistakes, but it does not validate that the collector will connect the event to existing datasets.

Close the client and flush transport workflush-client-transport

try:
    client.emit(event)
finally:
    flushed = client.close(timeout=10)
    if not flushed:
        raise RuntimeError('OpenLineage transport did not flush')

close uses -1 for an indefinite wait and zero for no timeout; bounded shutdown is safer for short-lived workers.

Alternatives

Package	Registry	Pick it when
acryl-datahub	PyPI	Your metadata platform is DataHub and you want its native emitters, ingestion recipes, and model
openmetadata-ingestion	PyPI	You run OpenMetadata and need its connector and ingestion framework rather than a protocol-only client
marquez-python	PyPI	You target Marquez-specific APIs and prefer its direct Python client