[
https://issues.apache.org/jira/browse/SPARK-59395?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Paul Sedra updated SPARK-59395:
-------------------------------
Description:
h1. Summary
Spark Declarative Pipelines (SDP) does not expose a reliable way to link an
individual flow execution to the Spark SQL executions and jobs it produces.
Pipeline-level attribution is possible, but flow-level attribution is not,
preventing external observability tools from reliably attributing Spark work to
a specific SDP flow.
h2. Current behavior
A caller can attach a pipeline-run identifier using Spark Connect metadata or
job tags, which allows Spark work to be associated with the containing pipeline
run. However, it does not identify which SDP flow or flow attempt produced that
work. The public {{StartRun}} API exposes no flow-execution identity, and tags
on the enclosing {{ExecutePlanRequest}} are not guaranteed to propagate into
asynchronously executed flows. Pipeline events expose status text and
timestamps, but no structured link between a flow and its Spark executions. SQL
text, query plans, and timestamps can only be used for inference, not
authoritative attribution.
h2. Minimal example
For two flows in the same pipeline run:
{code:java}
pipeline_run = R1
R1 → flow silver_orders → Spark job J1
R1 → flow gold_orders → Spark job J2
{code}
Current tagging can show:
{code:java}
R1 → J1
R1 → J2{code}
But it cannot deterministically show:
{code:java}
flow silver_orders → J1
flow gold_orders → J2{code}
h2. Runtime evidence
The OSS runtime knows the active flow at the relevant boundary:
GraphExecution.planAndStartFlow(flow) → FlowExecution.executeAsync → batch or
streaming execution. See [GraphExecution.scala|#L79] and
[FlowExecution.scala|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala#L141].
That internal context is not exposed as a supported public SDP callback,
protocol field, or Spark
execution-metadata contract.
h2. Expected behavior
Expose enough stable semantic identity for an external observer to determine:
- which logical SDP flow is executing;
- which individual execution or attempt is being observed; and
- which Spark SQL executions and/or Spark jobs belong to that execution,
allowing existing Spark
execution relationships to provide stage/task attribution.
The behavior should cover SDP batch and streaming flows. The implementation and
API shape are intentionally left to Spark maintainers; protocol metadata,
execution tags, structured events, or listener/event-log metadata are possible
mechanisms, not requirements.
h2. Acceptance criteria
# An SDP flow has an externally observable logical identity and an identity
for an individual execution or attempt.
# External tools can deterministically correlate that execution with its Spark
SQL executions and/or jobs, allowing existing Spark execution relationships to
provide stage/task attribution, without parsing SQL, plans, logs, or timestamps.
# Attribution remains correct for multiple flows and attempts, including batch
and streaming flows, while existing clients that do not use the new metadata
remain compatible.
h2. References
- SPARK-51727: SPIP: Declarative Pipelines
- SPARK-44591: Add jobTags to SparkListenerSQLExecutionStart
- SPARK-44612: Use jobTags in SparkListenerSQLExecutionStart to get SQL
Execution ID for Spark UI Connect page
-
[GraphExecution.scala|[https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala]|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala],
[FlowExecution.scala|[https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala]|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala],
and
[pipelines.proto|[https://github.com/apache/spark/blob/master/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto]|https://github.com/apache/spark/blob/master/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto]
was:
h1. Summary
Spark Declarative Pipelines (SDP) does not expose a reliable way to link an
individual flow execution to the Spark SQL executions and jobs it produces.
Pipeline-level attribution is possible, but flow-level attribution is not,
preventing external observability tools from reliably attributing Spark work to
a specific SDP flow.
h2. Current behavior
A caller can attach a pipeline-run identifier using Spark Connect metadata or
job tags, which allows Spark work to be associated with the containing pipeline
run. However, it does not identify which SDP flow or flow attempt produced that
work. The public {{StartRun}} API exposes no flow-execution identity, and tags
on the enclosing {{ExecutePlanRequest}} are not guaranteed to propagate into
asynchronously executed flows. Pipeline events expose status text and
timestamps, but no structured link between a flow and its Spark executions. SQL
text, query plans, and timestamps can only be used for inference, not
authoritative attribution.
h2. Minimal example
For two flows in the same pipeline run:
{code:java}
pipeline_run = R1
R1 → flow silver_orders → Spark job J1
R1 → flow gold_orders → Spark job J2
{code}
Current tagging can show:
{code:java}
R1 → J1
R1 → J2{code}
But it cannot deterministically show:
{code:java}
flow silver_orders → J1
flow gold_orders → J2{code}
h2. Runtime evidence
The OSS runtime knows the active flow at the relevant boundary:
GraphExecution.planAndStartFlow(flow) → FlowExecution.executeAsync → batch or
streaming execution. See [GraphExecution.scala|#L79] and
[FlowExecution.scala|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala#L141].
That internal context is not exposed as a supported public SDP callback,
protocol field, or Spark
execution-metadata contract.
h2. Expected behavior
Expose enough stable semantic identity for an external observer to determine:
- which logical SDP flow is executing;
- which individual execution or attempt is being observed; and
- which Spark SQL executions and/or Spark jobs belong to that execution,
allowing existing Spark
execution relationships to provide stage/task attribution.
The behavior should cover SDP batch and streaming flows. The implementation and
API shape are intentionally left to Spark maintainers; protocol metadata,
execution tags, structured events, or listener/event-log metadata are possible
mechanisms, not requirements.
h2. Acceptance criteria
# An SDP flow has an externally observable logical identity and an identity
for an individual execution or attempt.
# External tools can deterministically correlate that execution with its Spark
SQL executions and/or jobs, allowing existing Spark execution relationships to
provide stage/task attribution, without parsing SQL, plans, logs, or timestamps.
# Attribution remains correct for multiple flows and attempts, including batch
and streaming flows, while existing clients that do not use the new metadata
remain compatible.
h2. References
- SPARK-51727: SPIP: Declarative Pipelines
- SPARK-44591: Add jobTags to SparkListenerSQLExecutionStart
- SPARK-44612: Use jobTags in SparkListenerSQLExecutionStart to get SQL
Execution ID for Spark UI Connect page
-
[GraphExecution.scala|[https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala]|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala],
[FlowExecution.scala|[https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala]|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala],
and
[pipelines.proto|[https://github.com/apache/spark/blob/master/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto]|https://github.com/apache/spark/blob/master/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto]
> [SDP] Expose deterministic flow-to-Spark execution attribution
> --------------------------------------------------------------
>
> Key: SPARK-59395
> URL: https://issues.apache.org/jira/browse/SPARK-59395
> Project: Spark
> Issue Type: Improvement
> Components: Declarative Pipelines
> Affects Versions: 4.1.3
> Reporter: Paul Sedra
> Priority: Minor
>
> h1. Summary
> Spark Declarative Pipelines (SDP) does not expose a reliable way to link an
> individual flow execution to the Spark SQL executions and jobs it produces.
> Pipeline-level attribution is possible, but flow-level attribution is not,
> preventing external observability tools from reliably attributing Spark work
> to a specific SDP flow.
> h2. Current behavior
> A caller can attach a pipeline-run identifier using Spark Connect metadata or
> job tags, which allows Spark work to be associated with the containing
> pipeline run. However, it does not identify which SDP flow or flow attempt
> produced that work. The public {{StartRun}} API exposes no flow-execution
> identity, and tags on the enclosing {{ExecutePlanRequest}} are not guaranteed
> to propagate into asynchronously executed flows. Pipeline events expose
> status text and timestamps, but no structured link between a flow and its
> Spark executions. SQL text, query plans, and timestamps can only be used for
> inference, not authoritative attribution.
> h2. Minimal example
> For two flows in the same pipeline run:
> {code:java}
> pipeline_run = R1
> R1 → flow silver_orders → Spark job J1
> R1 → flow gold_orders → Spark job J2
> {code}
> Current tagging can show:
> {code:java}
> R1 → J1
> R1 → J2{code}
> But it cannot deterministically show:
> {code:java}
> flow silver_orders → J1
> flow gold_orders → J2{code}
> h2. Runtime evidence
> The OSS runtime knows the active flow at the relevant boundary:
> GraphExecution.planAndStartFlow(flow) → FlowExecution.executeAsync → batch or
> streaming execution. See [GraphExecution.scala|#L79] and
> [FlowExecution.scala|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala#L141].
> That internal context is not exposed as a supported public SDP callback,
> protocol field, or Spark
> execution-metadata contract.
> h2. Expected behavior
> Expose enough stable semantic identity for an external observer to determine:
> - which logical SDP flow is executing;
> - which individual execution or attempt is being observed; and
> - which Spark SQL executions and/or Spark jobs belong to that execution,
> allowing existing Spark
> execution relationships to provide stage/task attribution.
> The behavior should cover SDP batch and streaming flows. The implementation
> and API shape are intentionally left to Spark maintainers; protocol metadata,
> execution tags, structured events, or listener/event-log metadata are
> possible mechanisms, not requirements.
> h2. Acceptance criteria
> # An SDP flow has an externally observable logical identity and an identity
> for an individual execution or attempt.
> # External tools can deterministically correlate that execution with its
> Spark SQL executions and/or jobs, allowing existing Spark execution
> relationships to provide stage/task attribution, without parsing SQL, plans,
> logs, or timestamps.
> # Attribution remains correct for multiple flows and attempts, including
> batch and streaming flows, while existing clients that do not use the new
> metadata remain compatible.
> h2. References
> - SPARK-51727: SPIP: Declarative Pipelines
> - SPARK-44591: Add jobTags to SparkListenerSQLExecutionStart
> - SPARK-44612: Use jobTags in SparkListenerSQLExecutionStart to get SQL
> Execution ID for Spark UI Connect page
> -
> [GraphExecution.scala|[https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala]|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala],
>
> [FlowExecution.scala|[https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala]|https://github.com/apache/spark/blob/master/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala],
> and
> [pipelines.proto|[https://github.com/apache/spark/blob/master/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto]|https://github.com/apache/spark/blob/master/sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
