apache / apache/beam

[Feature Request]: Make lineage tracking pluggable

Open
#36,790 3 comments 0 reactions 1 assignee Claimed by @shnapz View on GitHub
java new feature P2
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
2d 2h
Merged PRs (30d)
205

Description

### What would you like to happen?

### Current lineage implementation

Currently, the lineage implementation is tightly coupled to Beam Metrics as the storage backend in both Java and Python SDK.

The Java static API returns two instances for source and sink:
```java
Lineage sources = Lineage.getSources();
Lineage sinks = Lineage.getSinks();
```
The `Lineage` instance provides different overloads of the `add()` method, which sends lineage data to metrics:
```java
if (MetricsFlag.lineageRollupEnabled()) {
((BoundedTrie) this.metric).add(segments);
} else {
((StringSet) this.metric).add(String.join("", segments));
}
```
Also `Lineage` class provides two public static methods to query Lineage from results:
```java
Set query(MetricResults results, Type type, String truncatedMarker)
Set query(MetricResults results, Type type)
```

### Requested functionality
Add a pluggable lineage tracking mechanism using ServiceLoader to decouple lineage reporting from core metrics infrastructure, enabling flexible observability without core changes. Scope: Java SDK (Python SDK in future work).

1. This change must preserve current public APIs
2. Minimize changes to I/O connectors that produce lineage; isolate changes to the `org.apache.beam.sdk.metrics.Lineage` class
3. Use a plugin approach via ServiceLoader discovery, following the existing pattern of `FileSystemRegistrar`. Key advantage is that registrars on classpath can read `PipelineOptions` and turn on or turn off depending on parameters. This satisfies the approach described in the [Open Lineage ticket](https://github.com/apache/beam/issues/33981):
```
options = PipelineOptions([
'--openlineage_enabled=true',
```
(note that OpenLineage is just a good example, it is out of scope of this request)
4. Do not provide any concrete plugin implementations. If no plugins are available, fall back to the existing metric-based lineage approach.
5. Unfortunately static `query` methods expose `MetricResults` as implementation detail. Leave them as is, so they are out of scope of this change.

### Deduplication

Native metrics like BoundedTrie/StringSet perform cross-worker consolidation automatically. In contrast, custom lineage shifts the responsibility for deduplication to the receiving system, whether that be OpenLineage or a custom-built implementation.

### Relationship to existing roadmap

~~This change will serve as a foundation for [[Feature Request]: Integrate Apache Beam with
Open Lineage](https://github.com/apache/beam/issues/33981) which is already put on the roadmap for Beam 3.0.~~
Existing [[Feature Request]: Integrate Apache Beam with
Open Lineage](https://github.com/apache/beam/issues/33981) attempts to add a different type of functionality to Beam 3.0 and does not overlap. So the current feature request does not interfere with it

### Testing strategy
- Unit tests for the plugin discovery mechanism
- Integration tests with mock lineage reporters
- Backward compatibility tests ensuring existing metric-based lineage still works when no plugins are present
- Cross-runner tests will initially focus on DirectRunner, with cross-runner compatibility expected to be inherited from the existing metrics infrastructure

### Documentation
- Update JavaDoc for `Lineage` class
- Add developer guide for implementing custom lineage reporters

### Issue Priority

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

### Issue Components

- [ ] Component: Python SDK
- [x] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.