cloud-fan commented on PR #56055: URL: https://github.com/apache/spark/pull/56055#issuecomment-4905948236
The spec is the right shape and answers the direction well — PSD as a binding contract (§2.1), the group as the failure/admission unit (§2.2), and activation by membership (§8, which removes the prototype's half-activated state). A few gaps before it's the implementation contract: **1. Cross-group scheduling / starvation (§4) — @mridulm's (b), and the weakest part.** §4 defines gang admission for one group but not how multiple groups (or a group vs. regular jobs) arbitrate — and a pipelined group holds its slots for the whole batch. Is admission FIFO? Can a large group be starved by a stream of small ones? When an admitted group has pinned its slots, does the next one wait or fail-fast (which means a second streaming query can't start on a busy cluster)? Are running groups' slots subtracted from "available" for the next check? **2. "Available slots" needs a precise definition (§4).** The prototype used `defaultParallelism()`, which returns `spark.default.parallelism`, not free slots — the barrier precedent uses `sc.maxNumConcurrentTasks(rp)` (`checkBarrierStageWithNumSlots`), which is the number to reuse. Also, `sum(numTasks)` ignores resource profiles: either state a group is single-resource-profile (and reject otherwise) or define per-profile accounting. **3. Output-commit vs. group-atomic failure (§5).** §5 runs output-commit immediately while deferring stage/job finish, so a result stage → task commits → sibling fails in the replay window → group reruns → double commit. RTM's sink is idempotent so RTM is safe, but as a generic primitive this needs either output-commit to also defer, or a stated requirement that in-group result-stage side effects be idempotent. **4. Drop the fetch-failure framing in §6 — state it as a mechanism.** Internal edges are streaming shuffle, so the base `FetchFailed`-means-resubmit concept doesn't apply inside a group; enumerating it implies a mechanism that doesn't exist. Better: *any member task failure, for any reason, fails the group; the base single-stage resubmit path is disabled for members* (that clause is what makes "all errors fail the group" actually true). Then note under §7 that a lost **external** durable input reruns the group rather than being recomputed in isolation — correct but coarser than base Spark, worth flagging as an intentional v1 simplification. **5. 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? **Minor:** (a) fail-fast admission → for streaming, crash-loops on restart; the barrier path retries `BarrierJobSlotsNumberCheckFailed` N times — say whether admission is retried or terminal. (b) "pipelined" over `StreamingShuffleDependency` is the better name (not streaming-specific) — worth a sentence on why. -- 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]
