88fantasy opened a new pull request, #4498:
URL: https://github.com/apache/streampark/pull/4498
## 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.address` is empty, so an existing deployment is
unaffected until an operator configures it — deliberate, since injecting
listener configuration for a jar the cluster's `lib/` does not have would break
job startup.
## Brief change log
**Configuration** — four new system settings (`lineage.gravitino.address`,
`.token`, `.namespace`, `lineage.flink.native.listener.enable`), a
`LineageConfig` bean, and a `lineage_enable` column on `t_flink_app` /
`t_spark_app` for per-application opt-in.
**Flink lineage extraction** — `FlinkSqlLineageExtractor` (one per shims
base, v1 and v2) plans the job's SQL in a throwaway `TableEnvironment` inside
the per-version shims classloader, reached through the existing
`FlinkShimsProxy` — the same mechanism `FlinkSqlServiceImpl.verifySql` already
uses. `CompiledPlanLineageParser` then walks the plan JSON from each
`dynamicTableSink` backwards over `edges` to the sources that reach it.
Two design points worth review:
- **Extraction runs in the console, not in the job.** StreamPark submits
application-mode jobs detached, so a `JobListener` registered inside the job
would never fire `onJobExecuted`, 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 the `FlinkStreaming`/`FlinkTable` contract and the
job runtime untouched.
- **Inputs are paired per sink, not flattened.** A `STATEMENT SET` job 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** — `DatasetIdentityRegistry` maps a connector's `WITH`
options to the `(namespace, name)` pair, and catalog-backed tables resolve from
the plan's qualified identifier. These 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. Unknown connectors are skipped with a WARN rather
than given a guessed generic fallback.
**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:
- Per-sink splitting produced exactly 2 edges, not 2×2.
- **No twin nodes**: the run added only the 2 new sink datasets; it reused
the existing source dataset rows written by another producer, which is the
whole point of the identity rules.
- START and COMPLETE both landed, both runs reached a terminal `COMPLETE`
state, and the graph generation advanced with `stale: false`.
- Run IDs recomputed independently matched the stored ones byte-for-byte.
- Fail-open confirmed both ways: an unknown connector (datagen -> print) and
an unreachable Gravitino address each left the job submitting and finishing
normally, with only WARN/ERROR logs.
Unit tests: `streampark-flink-shims-base` 27 tests,
`streampark-console-service` 115 tests, all passing. The regression guard for
the batch-mode point below was checked by reverting the fix and confirming the
test fails.
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_setting` rows on MySQL and PostgreSQL.
## Known limitations
- **Flink 2.x is not covered yet.** Extraction itself is version-agnostic,
but a Flink 2.x SQL job cannot currently be submitted on REMOTE mode at all
(the console bundles a fixed baseline Flink, and Flink resolves
`org.apache.flink.*` parent-first). That is a separate architectural issue, not
specific to lineage.
- **Table-level only**, no column-level lineage.
- The optional native `openlineage-flink` listener requires Flink 1.19+ and
the jar in the cluster's `lib/`; it is off unless the address is set.
- The PostgreSQL upgrade script covers `t_flink_app` only, because
`pgsql-schema.sql` defines 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
- Dependencies (does it add or upgrade a dependency): **yes**
- Adds `io.openlineage:openlineage-java` 1.29.0 (Apache-2.0).
- Upgrades `httpclient5` 5.1 -> 5.4.2 and pins `httpcore5`/`httpcore5-h2`
to 5.4.3, which that client requires. Without the explicit pin, an older
transitive `httpcore5` wins mediation and `httpclient5` fails at runtime with
`NoSuchMethodError`. Happy to split this out if maintainers would rather review
the HTTP client bump separately.
- Adds `jackson-databind` to `streampark-flink-shims-base` for parsing the
plan JSON.
- The public API: **no**
- The runtime per-record code paths (performance sensitive): **no** —
nothing is added to the job runtime; extraction happens once per submission in
the console.
- Anything that affects deployment: **yes** — new `t_setting` rows and a new
column on `t_flink_app`/`t_spark_app`, with upgrade scripts for both MySQL and
PostgreSQL.
## Documentation
- Does this pull request introduce a new feature? **yes**
- If yes, how is the feature documented? Configuration is surfaced in the
system settings UI with descriptions, and the classes carry javadoc covering
the design decisions above. Happy to add website documentation if maintainers
want it in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
