Skip to content

Add OpenLineage extension for Java SDK#39428

Open
rahul-madaan wants to merge 1 commit into
apache:masterfrom
rahul-madaan:openlineage-extension
Open

Add OpenLineage extension for Java SDK#39428
rahul-madaan wants to merge 1 commit into
apache:masterfrom
rahul-madaan:openlineage-extension

Conversation

@rahul-madaan

@rahul-madaan rahul-madaan commented Jul 22, 2026

Copy link
Copy Markdown

Adds sdks/java/extensions/openlineage, a new extension that emits OpenLineage events describing the datasets a Beam pipeline reads and writes, so OpenLineage consumers (Marquez, Atlan, and others) can build lineage graphs for Beam jobs on any runner.

addresses #39427

Design. The extension deliberately mirrors the architecture of the official OpenLineage Spark and Flink integrations so it is familiar to both communities:

  • Configuration extends io.openlineage.client.OpenLineageConfig (like SparkOpenLineageConfig/FlinkOpenLineageConfig) and resolves with the Spark integration's precedence: pipeline options > openlineage.yml > OPENLINEAGE__ env vars. Transport, disabled facets, dataset resolvers and circuit breaker reuse the standard openlineage-java config classes; no secrets flow through pipeline options.
  • OpenLineageRunner wraps any runner (--runner=OpenLineageRunner --openLineageDelegateRunner=FlinkRunner): extracts datasets from the pipeline graph via classpath-guarded per-IO visitors (PubsubIO typed, IcebergIO reflective since its getters are package-private; extensible through the public LineageProvider interface), mints a UUIDv7 run id at submission and propagates it via serialized options, emits START and terminal COMPLETE/ABORT/FAIL (with an errorMessage facet on failure), and supports a parent run facet for orchestrator-launched pipelines.
  • OpenLineageJobTracker emits periodic RUNNING events on the Flink integration's trackingIntervalInSeconds cadence (default 60s) and sweeps lineage reported through the Beam Lineage metrics API at each tick and at terminal state.
  • OpenLineageLineage implements the LineageBase plugin ([Feature Request]: Make lineage tracking pluggable #36790, --lineageType) to capture lineage live inside worker JVMs for long-running streaming jobs — where runner metrics are not readable mid-run — teeing every FQN back into the metrics store so existing metrics-based consumers keep working unchanged.
  • Dataset naming strictly follows the OpenLineage naming conventions. Iceberg datasets use the physical table location with the catalog table identity attached as a TABLE symlink, matching the Spark integration's IcebergHandler. Systems without a spec entry pass through with a conservative generic mapping.

Emission can never fail a pipeline: every hook swallows exceptions and emission runs inside the configured OpenLineage circuit breaker.

Relationship to prior work. This realizes the OpenLineage integration requested in #33981 (closed as not-planned) and builds directly on the pluggable lineage mechanism added in #36790OpenLineageLineage is a LineageBase implementation selected via --lineageType, i.e. the concrete consumer of that hook. It also incorporates the approach discussed on the dev@ thread of consuming the lineage metrics IOs already report and traversing the pipeline graph for source/sink pairs, without changing any IO. The two are combined: the --lineageType plugin captures lineage live in worker JVMs (needed for long-running streaming jobs), while the wrapper runner traverses the graph at submission and sweeps the Lineage metrics at terminal state (covers batch and Beam versions predating the plugin API). Granularity is job-level, matching the direction settled in #36790.

cc @charlespnh @shnapz — you have both been driving lineage / OpenLineage work in Beam; would appreciate your review or pointers here.

Testing. 26 unit tests: FQN-to-naming-convention conformance for every spec-listed system, graph extraction against real PubsubIO and real IcebergIO (Hadoop catalog with actual loadTable() location resolution, plus graceful fallback when a REST catalog is unreachable at submit time), end-to-end runner and plugin pipelines on the direct runner asserting the full emitted event sequence via a file transport, parent-facet and disabled-flag behavior, and config precedence. Also validated end-to-end against a live OpenLineage consumer: START/RUNNING/COMPLETE events from this module were ingested and produced the expected lineage assets. :sdks:java:extensions:openlineage:build (checkstyle, spotless, SpotBugs, javadoc, tests) and rat pass.

Dependencies. New dependency io.openlineage:openlineage-java (Apache-2.0), scoped to this module only. GCP and Iceberg IO modules are compile-only; their visitors activate at runtime only when the corresponding IO is already on the user's classpath.


Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

  • Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
  • Update CHANGES.md with noteworthy changes.
  • If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

Build python source distribution and wheels
Python tests
Java tests
Go tests

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

@gemini-code-assist

Copy link
Copy Markdown
Contributor

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

@rahul-madaan
rahul-madaan marked this pull request as draft July 22, 2026 06:29
Adds sdks/java/extensions/openlineage, which emits OpenLineage events
describing the datasets a Beam pipeline reads and writes, so lineage
backends (Marquez, Atlan, and other OpenLineage consumers) can build a
lineage graph for Beam jobs on any runner.

The extension follows the architecture of the official OpenLineage
Spark and Flink integrations:
- Configuration extends io.openlineage.client.OpenLineageConfig and is
  resolved with the same precedence as the Spark integration: pipeline
  options over openlineage.yml over OPENLINEAGE__ env vars. Transport,
  facets, dataset and circuit-breaker settings use the standard
  openlineage-java config classes.
- OpenLineageRunner wraps any runner, extracts datasets from the
  pipeline graph through classpath-guarded per-IO visitors (PubsubIO,
  IcebergIO; extensible via the public LineageProvider interface),
  mints a UUIDv7 run id at submission and propagates it via serialized
  options, and emits START and terminal COMPLETE/ABORT/FAIL events.
- OpenLineageJobTracker emits periodic RUNNING events on the Flink
  integration's trackingIntervalInSeconds cadence (default 60s) and
  sweeps lineage reported through the Beam Lineage metrics API.
- OpenLineageLineage implements the LineageBase plugin (--lineageType)
  to capture lineage live in worker JVMs for long-running streaming
  jobs, teeing FQNs back into the metrics store.
- Dataset naming strictly follows the OpenLineage naming conventions;
  Iceberg datasets use the physical table location with the catalog
  table identity attached as a TABLE symlink, matching the Spark
  integration's IcebergHandler.

Addresses apache#39427.
@rahul-madaan
rahul-madaan force-pushed the openlineage-extension branch from 77b409a to d1f1752 Compare July 22, 2026 07:42
@rahul-madaan
rahul-madaan marked this pull request as ready for review July 22, 2026 13:25
@gemini-code-assist

Copy link
Copy Markdown
Contributor

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

@github-actions

Copy link
Copy Markdown
Contributor

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant