apache / apache/beam

[Feature Request]: OpenLineage extension for the Java SDK

Open
#39,427 0 comments 0 reactions 0 assignees View on GitHub
java P2
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

### What would you like to happen?

Beam pipelines should be able to emit [OpenLineage](https://openlineage.io) events describing the datasets they read and write, so OpenLineage consumers (Marquez, Atlan, and others) can build lineage graphs for Beam jobs on any runner. This continues #33981 and builds on the pluggable lineage mechanism from #36790.

Proposed: a new `sdks/java/extensions/openlineage` module that follows the architecture of the official OpenLineage Spark and Flink integrations:

- Configuration extends `io.openlineage.client.OpenLineageConfig` and resolves with the Spark integration's precedence: pipeline options over `openlineage.yml` over `OPENLINEAGE__` environment variables. Transport, facets, dataset and circuit-breaker settings reuse the standard openlineage-java config classes.
- `OpenLineageRunner` wraps any runner (`--runner=OpenLineageRunner --openLineageDelegateRunner=FlinkRunner`), extracts datasets from the pipeline graph via classpath-guarded per-IO visitors (PubsubIO, IcebergIO; extensible through a 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.
- A job tracker 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 so runner-native consumers keep working.
- Dataset naming strictly follows the [OpenLineage naming conventions](https://openlineage.io/docs/spec/naming/); Iceberg datasets use the physical table location with the catalog identity as a TABLE symlink, matching the Spark integration's IcebergHandler.

Building on - https://github.com/apache/beam/issues/33981

### Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

### Issue Components

- [X] Component: Java SDK
- [ ] Component: Python SDK

Contributor guide

Open the contributing guide

Research direction

Start by reading the pluggable lineage mechanism in #36790 and the related work in #33981, then compare the proposed architecture with the OpenLineage Spark and Flink integrations. The proposed entry point is a new `sdks/java/extensions/openlineage` module. Done would mean providing the described runner and lineage plugin, dataset extraction and event reporting, with configuration and naming following the stated conventions.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, distributed-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.