[Console] Report Flink and Spark job lineage to Gravitino via OpenLineage - #4498
Open
88fantasy wants to merge 6 commits into
Open
[Console] Report Flink and Spark job lineage to Gravitino via OpenLineage#449888fantasy wants to merge 6 commits into
88fantasy wants to merge 6 commits into
Conversation
…eage
StreamPark currently emits no lineage at all, so jobs it submits are
invisible to a Gravitino lineage graph that other producers already write
into. This adds table-level lineage reporting for Flink SQL and Spark
applications, off unless configured.
Configuration (system settings, all new):
lineage.gravitino.address base URL; empty disables everything
lineage.gravitino.token bearer token, needed once Gravitino
authentication is enabled
lineage.gravitino.namespace OpenLineage namespace to report under
lineage.flink.native.listener.enable also inject the official
openlineage-flink listener config
Per-application opt-in via a new lineage_enable column on t_flink_app and
t_spark_app. Nothing is injected or emitted while the address is empty, so
an existing deployment is unaffected until it is configured — deliberate,
since injecting listener config for a jar the cluster's lib/ does not have
would break job startup.
Flink: the SQL text is planned in a throwaway TableEnvironment inside the
per-version shims classloader (via the existing FlinkShimsProxy, the same
mechanism SQL verification already uses), and the resulting CompiledPlan
JSON is walked from each sink backwards to the sources that reach it. The
console emits START on submission and COMPLETE/FAIL when the job reaches a
terminal state, so the lineage graph is only rebuilt for runs that actually
finished. Pairing inputs per sink rather than flattening them matters for
STATEMENT SET jobs: N independent INSERTs compile to N disconnected
subgraphs, and flattening would report N×M edges that do not exist.
Spark: OpenLineage's own listener is configured through appProperties,
leaving any key the user set explicitly untouched.
Dataset identity is resolved by connector (mysql-cdc, doris, catalog-backed
tables) into the exact (namespace, name) pair other producers into the same
Gravitino instance already use — a mismatch would split one physical table
into two nodes in the graph rather than fail visibly. An unrecognised
connector is skipped with a WARN instead of guessed at.
Everything on this path is fail-open: an unresolvable dataset, a plan that
will not compile, or an unreachable Gravitino logs and moves on. A lineage
gap must never fail a job submission.
Adds io.openlineage:openlineage-java (Apache-2.0) at 1.29.0. That client
needs httpclient5 5.4.x, so httpclient5 goes 5.1 -> 5.4.2 and httpcore5 is
pinned to 5.4.3 — without pinning, an older transitive httpcore5 wins
mediation and httpclient5 fails at runtime with NoSuchMethodError.
Schema changes ship as upgrade/mysql/3.0.0.sql and upgrade/pgsql/3.0.0.sql
alongside the install scripts. Both were verified by applying the previous
release's schema, running the upgrade, and diffing the result against a
fresh install: identical columns and t_setting rows on both databases. The
PostgreSQL script covers t_flink_app only, because pgsql-schema.sql defines
no Spark tables at all — a pre-existing gap this change does not widen.
FlinkSqlLineageExtractor exists twice — once per shims base — because the TableEnvironment bootstrap differs between Flink 1.x and 2.x. Only the 1.x copy had tests, so the batch-mode regression they guard against could reappear in the 2.x copy unnoticed. Mirrors the three shims-base cases against Flink 2.x, and widens doExtract to package-private there for the same reason as in the 1.x copy: the public entry point is fail-open and returns the same empty list whether the plan compiled or threw, which is exactly what the batch-mode test needs to tell apart.
…r-app switch reachable
Follow-up hardening on the Gravitino lineage feature.
Dataset identity no longer assumes one deployment's catalog. `resolveOne` used
to name every catalog-backed table `paimon://catalog/db`, which contradicted
both the PR's own "no guessed generic fallback" claim and
`DatasetIdentityRegistry`'s javadoc, and would silently misname Hive, JDBC or
Iceberg tables. The scheme now comes from the catalog's own `CREATE CATALOG ...
WITH ('type' = '...')`; a catalog attached outside the job's SQL has an
unknowable type and is skipped with a WARN, consistent with how an unknown
connector is already handled. Guessing is worse than reporting nothing here:
datasets deduplicate by exact string, so a wrong guess splits one physical
table into two graph nodes instead of failing loudly.
`SqlWithOptionsParser` gains two fixes that compounded with the above:
a qualified `CREATE TABLE db.tbl` captured the qualifier instead of the table
name, and `WITH (` was located without regard for string literals, so a column
`COMMENT 'see WITH (x)'` hijacked the options clause. Either one made a
connector-backed table look catalog-backed, which the old fallback then
mislabelled rather than dropped.
The per-application `lineageEnable` switch was only settable at creation time
through the REST API: `update()` and `copy()` in both manage services copy
updatable fields explicitly and did not carry it, and no UI control existed.
Adds the copies plus a Switch on the Flink and Spark application forms, with
submit wiring, edit-page backfill, types and zh/en messages.
Also: bound the pending-run map with a 30-day `expireAfterWrite` so runs whose
terminal state is never observed stop accumulating for the process lifetime;
skip plan edges pointing at absent nodes rather than NPE; and consolidate the
duplicated default namespace and `/api/lineage` endpoint onto `LineageConfig`.
BEHAVIOR CHANGE: `spark.openlineage.namespace` is now always injected,
defaulted to `streampark`, where it was previously injected only when
configured. An existing Spark install that left the namespace blank moves from
OpenLineage's own default to `streampark`. This makes Spark match the Flink
path, so both engines in one StreamPark install report to the same namespace.
streampark-flink-shims-base 33 tests and streampark-console-service 115 tests
pass; six new tests cover catalog-type derivation, unknown-catalog skipping,
malformed plan edges, qualified table names, WITH inside a literal, and
CREATE CATALOG parsing.
Fixes all 16 new-code issues SonarCloud reported for apache#4498. One of them was a real defect rather than a smell: the WITH-option literal patterns repeated a group greedily, which the JDK matches by recursing once per character, so a long option value (an inline certificate, a serialized properties blob) threw StackOverflowError on the submission path instead of yielding lineage. The repetitions are now possessive, which is matched iteratively; nothing inside a literal can consume the quote that closes it, so a well-formed entry never needed the backtracking that possessive matching gives up. The rest are maintainability fixes with no behaviour change: - Narrow the identifier patterns' CASE_INSENSITIVE flag to an inline (?i:...) around the keywords alone, which both removes the duplicate character ranges and keeps identifier matching case-exact. - Split CompiledPlanLineageParser.parse into per-step methods, bringing its cognitive complexity from 33 to within the limit and dropping the extra loop exits, and name the repeated plan JSON field literals. - Extract the per-pipeline emit out of GravitinoLineageServiceImpl's two nested try blocks, which also merges the duplicated START and terminal emit loops into one. - Catch Exception | LinkageError instead of Throwable when extracting lineage: the reflective call crosses into a shims classloader built for another Flink version, where a missing class surfaces as NoClassDefFoundError, so that case must stay fail-open.
…d name Second round of SonarCloud findings, all introduced by the previous commit's regex rework and all in SqlWithOptionsParser. The shared IDENTIFIER fragment carried its own (?:...) so that it could hold an alternation, which meant every capture group built from it wrapped a group that was already grouped (S6395). It is now a bare alternation and each use wraps it in the group kind it actually needs. Keeping the CREATE keywords and the declared name in one pattern also put the identifier alternation in there twice, which is most of why the table pattern scored 27 against the 20 allowed (S5843). They are now separate patterns, matched in sequence and anchored at the keywords' end, so a declared name has one definition shared by both statement kinds instead of one inlined copy each. No behaviour change: same statements match, same local name is captured.
…of grouping the flag Third and last round of SonarCloud findings on this path. The inline (?i:...) ran to the end of its pattern, so the group had no effect of its own (S6395) while still counting as a nesting level for each of the nine whitespace quantifiers inside it, which is where the keyword pattern's complexity of 22 came from (S5843). Now that the keywords are a pattern of their own, they can simply be compiled CASE_INSENSITIVE: what forced the inline form in the first place was the identifier character classes, which turn into duplicate ranges under that flag, and those live in DECLARED_NAME alone.
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
StreamPark emits no data lineage today. This adds table-level lineage reporting for Flink SQL and Spark applications as OpenLineage run events, sent to an Apache Gravitino server's
POST /api/lineage.Closes #4495
Disabled unless configured. Nothing is injected and nothing is emitted while
lineage.gravitino.addressis empty, so an existing deployment is unaffected until an operator configures it — deliberate, since injecting listener configuration for a jar the cluster'slib/does not have would break job startup.Depends on #4496 and #4500
This PR is self-contained and merges cleanly on its own, but a Flink SQL job cannot be submitted at all on current
dev, so lineage cannot be observed end-to-end without these two:[Flink] Fix Flink SQL job submission failures on the client path[Flink] Make Flink 2.x job submission work on REMOTE mode(fixes the shims classloader matching thate770d2e8ebroke)Measured, not assumed: deploying a build of this branch alone and starting the Flink SQL verification job fails with
before the job ever reaches the cluster. Both PRs merge into this branch without conflict, and that merged combination is what the end-to-end verification below was carried out on.
Worth noting what that run showed about this PR's own failure policy: lineage extraction ran, resolved every dataset correctly, and the submission failure that followed was entirely unrelated to it — the fail-open contract held in exactly the situation it exists for. Review order does not matter; merge order does.
Brief change log
Configuration — four new system settings (
lineage.gravitino.address,.token,.namespace,lineage.flink.native.listener.enable), aLineageConfigbean, and alineage_enablecolumn ont_flink_app/t_spark_appfor per-application opt-in, surfaced as a switch on the Flink and Spark application forms.Flink lineage extraction —
FlinkSqlLineageExtractor(one per shims base, v1 and v2) plans the job's SQL in a throwawayTableEnvironmentinside the per-version shims classloader, reached through the existingFlinkShimsProxy— the same mechanismFlinkSqlServiceImpl.verifySqlalready uses.CompiledPlanLineageParserthen walks the plan JSON from eachdynamicTableSinkbackwards overedgesto the sources that reach it.Two design points worth review:
JobListenerregistered inside the job would never fireonJobExecuted, and Gravitino only promotes a run to the current topology on COMPLETE — a START-only stream never updates the graph. Emitting all three events from the console's own state machine keeps the lifecycle correct and leaves theFlinkStreaming/FlinkTablecontract and the job runtime untouched.STATEMENT SETjob with N independent INSERTs compiles to N disconnected subgraphs; flattening the plan into a single input/output set would report N×M edges that do not exist.Dataset identity — these
(namespace, name)strings must match other producers into the same Gravitino instance byte-for-byte: datasets are deduplicated by exact string, so a mismatch does not fail loudly, it silently splits one physical table into two graph nodes. Two routes, and deliberately no third:CREATE TABLE ... WITH (...)carries its physical location in its connector options, soDatasetIdentityRegistrydecides;'type'—CREATE CATALOG c WITH ('type' = 'paimon', ...)yieldspaimon://c/db, a Hive catalog yieldshive://c/db.An unknown connector, or a catalog attached outside the job's SQL whose type is therefore unknowable, is skipped with a WARN rather than given a guessed fallback. Guessing is worse than reporting nothing here, for the deduplication reason above.
Spark lineage — OpenLineage's Spark listener is configured through
appProperties, never overriding a key the user set explicitly.Fail-open throughout — an unresolvable dataset, a plan that will not compile, an unreachable Gravitino: each logs and continues. A lineage gap must never fail a job submission.
Verifying this change
End-to-end against a real Flink 1.20.4 standalone cluster, with a two-sink mysql-cdc -> Paimon job, verified directly against the lineage backend's own tables rather than StreamPark's logs:
COMPLETEstate, and the graph generation advanced withstale: false.Extraction was additionally re-checked against both shims bases by running the console's own
FlinkShimsProxycode path in a separate JVM, against the real Flink Home of each cluster and the real job SQL: Flink 1.20.4 and Flink 2.2.1 both resolve the same two pipelines, and the resulting identities match the rows already present in Gravitino, so a re-run reuses them instead of creating twins.Unit tests:
streampark-flink-shims-base33 tests,streampark-console-service115 tests, all passing.Both database upgrade scripts were verified by applying the previous release's schema, running the upgrade, and diffing against a fresh install — identical columns and
t_settingrows on MySQL and PostgreSQL.Behavior note for existing installs
spark.openlineage.namespaceis now always injected, defaulted tostreampark, where an earlier revision of this PR injected it only when an operator had configured one. A Spark install that leaves the namespace blank therefore moves from OpenLineage's own default tostreampark. This makes Spark match the Flink path, so both engines in one StreamPark install report under the same namespace rather than two.Known limitations
openlineage-flinklistener requires Flink 1.19+ and the jar in the cluster'slib/; it is off unless the address is set.DatasetIdentityRegistrycurrently carries rules formysql-cdcanddoris. Other connectors are skipped with a WARN until a rule is added; the registry is a singleswitchand each rule is a few lines.t_flink_apponly, becausepgsql-schema.sqldefines no Spark tables at all — a pre-existing gap this PR neither widens nor fixes.Does this pull request potentially affect one of the following parts
io.openlineage:openlineage-java1.29.0 (Apache-2.0).jackson-databindtostreampark-flink-shims-basefor parsing the plan JSON.httpclient55.1 -> 5.4.2 and pinshttpcore5/httpcore5-h2to 5.4.3, which that client requires. Without the explicit pin, an older transitivehttpcore5wins mediation andhttpclient5fails at runtime withNoSuchMethodError. This is the widest-blast-radius part of the PR, so the call sites it affects, for reviewers who want to scope it:HttpClientUtils,YarnUtils,FlinkCluster,FlinkAppHttpWatcher,FlinkClusterWatcher,FlinkSessionSubmitHelper,FlinkCheckpointWatcher,FlinkJobStatusWatcher,FlinkMetricWatcher. Happy to split it into its own PR if maintainers would rather review the HTTP client bump separately.t_settingrows and a new column ont_flink_app/t_spark_app, with upgrade scripts for both MySQL and PostgreSQL.Documentation