Repository navigation
Add OpenLineage extension for Java SDK - #39428
rahul-madaan wants to merge 10 commits into
Conversation
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
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.
77b409a to
d1f1752
Compare
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
assign set of reviewers |
|
Assigning reviewers: R: @Abacn for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
|
cc: @charlespnh @shnapz since you have both been driving lineage / OpenLineage work in Beam; would appreciate your review or pointers here. |
Managed.write(ICEBERG) wraps the destination in PortableIcebergDestinations
even when the configured table is a constant, so the IcebergIO.WriteRows
node carries no static TableIdentifier and graph extraction skipped it.
Recover the table name from the destination template when its
RowStringInterpolator has no per-record placeholders; genuinely dynamic
templates (e.g. "db.table_{field}") are still skipped, as those
destinations are only knowable at runtime.
Addresses apache#39427.
- Managed Iceberg write through OpenLineageRunner: asserts the START event carries the physical-location dataset with the catalog TABLE symlink after a real write to a Hadoop-catalog table. - Unbounded pipeline with non-blocking run: asserts periodic RUNNING heartbeats at the configured tracking interval and the CANCELLED to ABORT terminal mapping on cancel(). - Failing pipeline: asserts FAIL is emitted with an errorMessage run facet carrying the user exception. Addresses apache#39427.
4d7ae6b to
feae18e
Compare
Fixes issues found in an end-to-end review of the extension: - Context construction can no longer throw: malformed run ids fall back to the deterministic id, malformed parent run ids omit the facet, and invalid transport or circuit-breaker configuration degrades to a non-emitting client with an error logged. The LineageBase plugin is instantiated reflectively during worker initialization, so a configuration mistake previously crash-looped the job. - The plugin no longer freezes the run id before OpenLineageRunner has minted it: the context is re-resolved at submission and the plugin looks the context up per call, so driver and worker events always share one run id. - The terminal event is now emitted deterministically through the returned PipelineResult wrapper on waitUntilFinish()/cancel(), instead of relying on the periodic tracker poll racing JVM exit. UPDATED now maps to COMPLETE rather than FAIL. - A finished context is replaced on the next execution so reused worker JVMs (e.g. Flink session clusters) track the new run. - PubsubIO extraction guards each ValueProvider access independently, so a runtime-only topic no longer discards an accessible subscription in templated pipelines. - Iceberg extraction falls back to the catalog identity when a table location cannot be parsed as a URI instead of dropping the dataset. - The metrics tee now populates both the StringSet and BoundedTrie forms so Lineage.query keeps working under enable_lineage_rollup. - Azure datasets use the wasbs blob endpoint and HDFS paths keep their leading slash, matching openlineage-java's canonical naming. - The tracker is not started when lineage is disabled, non-positive tracking intervals fall back to the default, the client is closed after the terminal event, and the dataset dedup key can no longer collide across namespace/name boundaries. Addresses apache#39427.
feae18e to
3ee033b
Compare
|
Reminder, please take a look at this pr: @Abacn |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @chamikaramj for label java. Available commands:
|
|
Reminder, please take a look at this pr: @chamikaramj |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @Abacn for label java. Available commands:
|
|
Reminder, please take a look at this pr: @Abacn |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @chamikaramj for label java. Available commands:
|
|
Reminder, please take a look at this pr: @chamikaramj |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @Abacn for label java. Available commands:
|
The iceberg module ended Java 11 support in Iceberg 1.11.0, so the extension can no longer resolve it while declaring Java 11.
Two threads can observe pipeline completion: the periodic job tracker daemon and whichever thread calls waitUntilFinish() or cancel(). The context claimed the terminal transition with a compare-and-set before emitting, so when the tracker daemon won the claim and the JVM exited part way through emission, the other thread found the run already claimed and emitted nothing. The run then stayed open in the backend forever. This reproduced consistently with the console transport on a portable runner, where emission is slow enough to lose the race. onJobFinished now holds the instance monitor across emission and marks the run finished only once the transport has accepted the event, so a second thread waits for delivery instead of returning early, which also keeps the JVM alive until the event is out. A rejected event leaves the run unsettled so a later completion signal retries it, rather than silently dropping the terminal event on a transient transport failure. The deterministic path also stops the tracker before finishing, so the two are not racing in the first place. EventEmitter.emit now reports whether the transport took the event.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #39428 +/- ##
============================================
- Coverage 58.52% 58.45% -0.07%
- Complexity 5980 13733 +7753
============================================
Files 1459 2593 +1134
Lines 208448 269751 +61303
Branches 5364 11178 +5814
============================================
+ Hits 121985 157682 +35697
- Misses 82166 106027 +23861
- Partials 4297 6042 +1745
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Apache Rat audits every file in the tree, including resources. Service files take '#' comments, and sdks/java/core carries the same header on its own test service file.
|
Reminder, please take a look at this pr: @Abacn |
|
It looks like this has bounced around for a while without review. One reason for this is probably that it is so large - 3,600 lines is quite a lot for a reviewer to manage. @rahul-madaan could you please split this up into several smaller PRs? I'd probably also recommend writing up a short design document and sharing this with Beam's dev list (https://beam.apache.org/community/contact-us/). Jumping straight into this much code will be challenging otherwise |
|
waiting on author |
github-actions
Bot
added
slow-review
Next Action: Author
and removed
slow-review
Next Action: Reviewers
labels
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:
io.openlineage.client.OpenLineageConfig(likeSparkOpenLineageConfig/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.OpenLineageRunnerwraps 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 publicLineageProviderinterface), mints a UUIDv7 run id at submission and propagates it via serialized options, emits START and terminal COMPLETE/ABORT/FAIL (with anerrorMessagefacet on failure), and supports a parent run facet for orchestrator-launched pipelines.OpenLineageJobTrackeremits periodic RUNNING events on the Flink integration'strackingIntervalInSecondscadence (default 60s) and sweeps lineage reported through the BeamLineagemetrics API at each tick and at terminal state.OpenLineageLineageimplements theLineageBaseplugin ([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.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 #36790 —
OpenLineageLineageis aLineageBaseimplementation 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--lineageTypeplugin captures lineage live in worker JVMs (needed for long-running streaming jobs), while the wrapper runner traverses the graph at submission and sweeps theLineagemetrics 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) andratpass.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:
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, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.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)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.