Search before asking
Description
StreamPark emits no data lineage of any kind — lineage and openlineage have zero hits across the repository. Where a lineage catalogue already exists alongside StreamPark, jobs submitted through StreamPark are the blind spot in it: everything submitted by other means shows up in the graph, and StreamPark's jobs do not.
This proposes reporting table-level lineage for Flink SQL and Spark applications as OpenLineage run events, to an Apache Gravitino server's POST /api/lineage endpoint.
Shape of the proposal:
- Configuration lives in system settings: a Gravitino address, an auth token, an OpenLineage namespace, and a toggle for the official
openlineage-flink listener. Reporting is opt-in per application via a new lineage_enable column on t_flink_app / t_spark_app.
- Nothing happens until the address is configured. No config injection, no emission. Injecting listener configuration for a jar that a cluster's
lib/ does not have would break job startup on existing deployments, so an unconfigured install must be byte-for-byte unaffected.
- Flink lineage is extracted in the console, not in the job. The SQL is planned in a throwaway
TableEnvironment inside the per-version shims classloader (through the existing FlinkShimsProxy, the same mechanism SQL verification already uses) and the resulting CompiledPlan JSON is walked from each sink backwards to the sources reaching it. This matters: StreamPark submits YARN/K8s application-mode jobs detached, so a JobListener registered inside the job would never call back, and a lineage backend that only ever receives START never rebuilds its graph. Emitting START on submission and COMPLETE/FAIL from the console's own state machine keeps the lifecycle correct without touching the FlinkStreaming/FlinkTable contract or the runtime at all.
- Inputs are paired per sink, not flattened. A
STATEMENT SET job with N independent INSERTs compiles to N disconnected subgraphs; flattening the whole plan into one input/output set would report N×M edges that do not exist.
- Spark uses OpenLineage's own Spark listener, configured through
appProperties, without overriding any key the user set explicitly.
- Everything on this path is fail-open. An unrecognised connector, a plan that will not compile, an unreachable lineage server — each logs and moves on. A lineage gap must never fail a job submission.
One design point worth review: dataset identity must be byte-identical to what other producers writing into the same lineage graph use for the same physical table, since datasets are deduplicated by exact (namespace, name) string. A mismatch does not fail loudly — it silently splits one physical table into two nodes. The proposal therefore resolves identity per connector against a small explicit registry and skips unknown connectors with a WARN rather than guessing a generic fallback.
Usage Scenario
A platform team runs a lineage catalogue that already ingests lineage from jobs submitted outside StreamPark. They want the Flink SQL and Spark jobs their users submit through StreamPark to appear in the same graph, merged with the existing nodes rather than duplicated beside them, without asking users to change how they write SQL and without any change to jobs already in production.
Related issues
No response
Are you willing to submit a PR?
Code of Conduct
Search before asking
Description
StreamPark emits no data lineage of any kind —
lineageandopenlineagehave zero hits across the repository. Where a lineage catalogue already exists alongside StreamPark, jobs submitted through StreamPark are the blind spot in it: everything submitted by other means shows up in the graph, and StreamPark's jobs do not.This proposes reporting table-level lineage for Flink SQL and Spark applications as OpenLineage run events, to an Apache Gravitino server's
POST /api/lineageendpoint.Shape of the proposal:
openlineage-flinklistener. Reporting is opt-in per application via a newlineage_enablecolumn ont_flink_app/t_spark_app.lib/does not have would break job startup on existing deployments, so an unconfigured install must be byte-for-byte unaffected.TableEnvironmentinside the per-version shims classloader (through the existingFlinkShimsProxy, the same mechanism SQL verification already uses) and the resultingCompiledPlanJSON is walked from each sink backwards to the sources reaching it. This matters: StreamPark submits YARN/K8s application-mode jobs detached, so aJobListenerregistered inside the job would never call back, and a lineage backend that only ever receives START never rebuilds its graph. Emitting START on submission and COMPLETE/FAIL from the console's own state machine keeps the lifecycle correct without touching theFlinkStreaming/FlinkTablecontract or the runtime at all.STATEMENT SETjob with N independent INSERTs compiles to N disconnected subgraphs; flattening the whole plan into one input/output set would report N×M edges that do not exist.appProperties, without overriding any key the user set explicitly.One design point worth review: dataset identity must be byte-identical to what other producers writing into the same lineage graph use for the same physical table, since datasets are deduplicated by exact
(namespace, name)string. A mismatch does not fail loudly — it silently splits one physical table into two nodes. The proposal therefore resolves identity per connector against a small explicit registry and skips unknown connectors with a WARN rather than guessing a generic fallback.Usage Scenario
A platform team runs a lineage catalogue that already ingests lineage from jobs submitted outside StreamPark. They want the Flink SQL and Spark jobs their users submit through StreamPark to appear in the same graph, merged with the existing nodes rather than duplicated beside them, without asking users to change how they write SQL and without any change to jobs already in production.
Related issues
No response
Are you willing to submit a PR?
Code of Conduct