mridulm commented on PR #57092: URL: https://github.com/apache/spark/pull/57092#issuecomment-4933619232
Thanks for the response - the rest look good to me. A few follow ups: > > In general, shuffle is one to many (1 producer, many consumers) - are we restricting/enforcing it to 1:1 ? If no - what is the behavior if/when it is used across jobs (DAGs) ? > > Short answer, Yes. > > You're right that shuffle is 1:many in general. Worth separating where that fan-out actually comes from: a pipelined edge only ever originates from the physical-planning rule marking an exchange, and one ShuffleDependency gets more than one consumer only via exchange/stage reuse — ReuseExchangeAndSubquery (self-join, shared subquery/CTE, self-union) or submitMapStage. Branching alone doesn't do it: without the reuse rule, two identical subtrees plan to two separate ShuffleExchangeExec instances, each with its own shuffleDependency (it's a lazy val per node), i.e. two independent 1:1 shuffles. So "fan-out of a pipelined edge" ≡ "a pipelined dependency got reused/shared." > > That makes fan-out fundamentally incompatible with a transient incremental shuffle, not just unimplemented: reuse relies on a durable, independently-fetchable materialized output that N consumers each read on their own schedule. A transient shuffle is a live, once-through stream from a still-running producer — there's no stored copy to hand to a second consumer, and concurrent readers would need fan-out-aware backpressure that doesn't exist. Fan-out becomes well-defined only over a persistent/replayable medium (e.g. a Kafka-backed shuffle, where consumers replay from offsets) — which is the persistentHint axis the spec already lists as a non-goal and defers. So 1:many isn't off the table forever; it's gated on that persistent-shuffle capability, which isn't on the table now. > We should not assume PG/streaming shuffle is used only from sql/RTM - the constructs can be directly depended on - and I would expect some interesting usecases to develop as well. > On how we do it? Single-ownership and 1:1 are the same mechanism — a pipelined ShuffleDependency is never reused or shared. A group is rejected fail-fast if a pipelined producer has more than one consuming stage or its stage carries more than one jobId. That single rule delivers both no-fan-out and no-cross-job-sharing. This also answers your later point that, from the scheduler's perspective, reuse can happen unless explicitly prevented — agreed, so we prevent it explicitly rather than relying on a caller (e.g. a new micro-batch minting fresh shuffleIds) to avoid it incidentally. Shuffle dependency is tied to the DAG, while stage is tied to a specific job execution of the dag. Making streaming shuffle id tied to a a job execution, as proposed, should have interesting implications - I have not thought through it - but it is good to call this out explicitly. > > "Cached/persisted RDD in a member's within-stage chain" -> how we do enforce this ? > > Enforced at stage/group creation: we walk each member stage's within-stage RDD chain — the narrow-dependency lineage, stopping at shuffle boundaries — and reject if any RDD has a non-NONE storage level (RDD.getStorageLevel). It reuses the DAGScheduler's existing within-stage traversal, so it's the same primitive the scheduler already uses, not new machinery. The shuffle-boundary stop is what makes the scope right: a cached complete input reached across a materialized shuffle (e.g. the static side of a stream-static join, or a broadcast) is outside the within-stage chain and correctly not flagged — only a cached RDD inside a member's own stage, which would freeze partial incremental output, is rejected. > > TBH, its is safe to use a cached/persisted RDD in a PG but not sure if it would ever be useful especially since structured streaming queries don't such caching. I guess if you want to reuse the results from an RDD part of the PG in another spark job? I will drop this from the incompatibility matrix for now. This enforcement sounds good to me. Note - there might be similar interaction with checkpoint as well, that would need to be addressed. > > > Just because at submission time there were insufficient resources to run it, does not mean that will continue to be the case. See existing barrier mode for insights. I believe this is what Wenchen is also driving at as well. > > So my proposal here in the spec at least for v1 is that any retries are left to the caller. There is no built-in retry or queuing mechanism. I think that is good enough for v1. We can always augment the spec later to include retries. This would not work except for simple scenarios (linear chain of prefix* -> PG -> suffix*) - users would need to provision nontrivial resources to ensure there is sufficient capacity to run all stages which can run concurrently with a PG (including other PGs), and the PG itself. Given this is already support for barrier scheduling, it should be possible to adapt for PG as well ? > > > Output-commit > > > I am concerned about the formulation - commit handling tends to be tricky. For the case when ResultStage is part of > PG : why cant we not have similar behavior as what currently exists ? > > (I am assuming this only applies when result stage is part of the PG - not side channel writes) > > Let me clarify this. There is actually no change to the existing output-commit path. The only genuinely new integration point is state cleanup: a member whose tasks all succeeded leaves a fully-populated authorizedCommitters array (the coordinator only clears a slot when the holder fails), so on group teardown we must drop each member's coordinator state — stageEnd / fresh stage ids on rerun — or the rerun's commits would be denied against the dead attempt's holders. commit state is only for result stage ? And result stage is what determines job termination ? In case of PG as well, if result stage terminates - I would expect all other stages to get torn down ? In other words, cleanup is triggered by result stage completion -> which is what happens today as well ? I am trying to see what I am missing here - I dont know enough about streaming nuances to know if there are any sub cases I might be missing ! -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
