jerrypeng commented on PR #57092: URL: https://github.com/apache/spark/pull/57092#issuecomment-4909880483
@cloud-fan thank you for your detail feedback. I will address them inline. > 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? Capacity is measured against free slots (running groups' and regular jobs' tasks subtracted from maxNumConcurrentTasks). A group that doesn't fit fails its admission attempt rather than queuing — no waiting queue, no partial reservation, so it matches barrier and can't deadlock on slots a sibling holds. I think this is good enough for now. In the future, we can consider if we want to augment the spec with wait / queuing capabilities > "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. Switched the definition to the cluster's concurrent-task capacity — the value barrier's slot check uses (sc.maxNumConcurrentTasks(rp)) — explicitly not defaultParallelism(). On resource profiles: v1 requires a group to be single-profile and fail-fast rejects a mixed-profile group; per-profile accounting is a documented follow-up (the RTM shapes are single-profile). > 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. I will clarify this. The current proposal simply requires an in-group result stage's side effects to be idempotent — the standard streaming model, where a batch is re-delivered on recovery and the sink absorbs it; a group rerun is the same re-delivery. Deferring the commit to group completion is noted as a future augmentation if a non-idempotent committer ever needs support. > 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. I will clarify / refine this a bit more > 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? This can't happen across micro-batches: each micro-batch is a new job built from a fresh query plan (a new IncrementalExecution), whose exchanges mint new ShuffleDependency objects with new shuffleIds. Since shuffleIdToMapStage reuse is keyed by shuffleId, a new batch never matches — and so never binds to — a prior batch's stage. It's a consequence of existing Spark behavior rather than something the spec needs to add. > 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. Clarified in the doc. -- 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]
