jerrypeng commented on PR #57092: URL: https://github.com/apache/spark/pull/57092#issuecomment-4922188167
@mridulm thank you for taking the time to review this even when you are on vacation! Let me respond to your comments in line: > Can we call out that in a DAG, a PG can have multiple "output" (regular shuffle) when ResultStage is not part of it ? Yup, this is and should be supported, will clarify. > Also, just as with barrier stage - a submitted DAG could have multiple pipelined group within it - with these PG's potentially running concurrently. Yup, will explicitly mention. > 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. 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. > Teardown is by group membership, not producer availability. > We do track shuffle loss when there is executor loss - should be possible to plumb that into PG as well. > Btw, this might be more strict than it needs to be ? > For example, if all output from a producer has been consumed successfully, and then the executor/node goes down -> does not actually impact the DAG ? would fetch failure be a better way to identify this failure mode ? > It is fine to start with a stronger formulation for simplicity btw ! What you called out make sense, though I think it is an optimization we may not need to do for v1. The fail fast mechanism should be good enough for now. In regards to using FetchFailed errors to signal this scenario. The streaming reader should reuse the FetchFailed channel as the signal to the scheduler, rather than the scheduler inferring failure from executor loss. Executor loss is the over-eager signal — it fires even when the producer had finished and its output was fully consumed, so it can't tell a harmful loss from a harmless one. The consumer can: it's the one reading, so it knows whether a producer's disappearance actually cost it data. Having the reader raise FetchFailed when (and only when) a read genuinely fails makes the consumer's real experience the trigger — a post-consumption producer loss produces no fetch failure and no teardown, which is exactly the precision you're after. The one adaptation: a FetchFailed on a pipelined edge must route to group failure, not the base single-stage mapper-resubmit (which is disabled for group members, and which is itself the source of the resubmit-then-hang problem). So we reuse the FetchFailed signal and its plumbing, but the handler recognizes the pipelined edge and reruns the group rather than recomputing the one mapper. Recovery stays group-atomic (the transient edge can't be re-fetched); only the detection moves from "driver infers from executor loss" to "consumer reports via FetchFailed." I will add a section in the spec to describe this as an optimization post-v1 > "Regular shuffle internal to a group — fail-fast / unsupported." -> I did not understand why this needs to fail. we would simply split the DAG into muliple PG ? Yes, you are right. That was mine intention but perhaps this bullet point added confusion. The intention was simply to indicate this scenario i.e. a regular shuffle being part of a PG is not possible. In reality, like you mentioned and what is mentioned in section 3, it will just split the DAG into multipe PGs. Though we should implement a check to make sure such a invariants hold. > Section 9: "Push-based shuffle merge" -> can you clarify this why this cant be supported for input and/or output ? Section 9 describes unsupported interactions with existing mechanism **within** a PG So for "Push-based shuffle merge" it simply means it cannot be used as an incremental shuffle in a PG. Of course it can be used as input or output for a PG just like a regular shuffle as you mentioned. > Section 9 "Statically-indeterminate producer" and "Checksum-mismatch" -> I did not understand this : a child of PG could have failed when reading PG output, and so requires PG reexecution - and so might be impacted by (INDETERMINATE) parent. Yes, it is simply describing the incapability of these mechanism to exist within a PG. It should work as it should outside of a PG. > "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. > 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. > Slot check > nit: This needs to factor in the stage requirements (resource profile/cores per task) * num partitions, across all stages in the group. Will make that more clear. > 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. > These are implementation details of a specific usecase. > From scheduler perspective, you can have reuse - unless we explicitly prevent/enforce it. I think you are referring a response of mine to @cloud-fan question: "Cross-job stage reuse (§3/§4). Each micro-batch is a new job, but the base scheduler reuses shuffle-map stages via shuffleIdToMapStage. What forces a pipelined edge to create a fresh stage rather than binding batch N to batch N-1's cached stage?" We are talking about the structured streaming use case there. Though I think you question is more general about exchange / shuffle re-use? If so, that this not support in a PG since the assumption currently is that the data in an incremental shuffle is transient thus cannot be re-used. -- 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]
