jerrypeng commented on PR #57092: URL: https://github.com/apache/spark/pull/57092#issuecomment-4942823045
@mridulm thank you for your feedback! Responding in line > 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. Agreed on the framing, and working through it changed my answer on fan-out — for the better. Two things: On generality: you're right that PG / pipelined shuffle is a generic DAGScheduler construct and I shouldn't justify its properties via SQL/RTM specifics (ReuseExchangeAndSubquery, etc.). On fan-out — I'm walking back "incompatible." 1:N (one producer, many consumers) is a supported model; v1 just defers it. The mechanism differs from a regular shuffle, but the model is the same. Concretely, a consumer's PG membership picks its transport: - In-PG consumers read the incremental edge — a live stream, so they must be co-scheduled with the producer (all live at once). This needs multicast (one producer pushing each record to N reader channels) plus per-consumer backpressure; that isn't built in the streaming shuffle today, but it isn't fundamentally hard either. Deferred. - Out-of-PG consumers read a materialized edge and fetch it by coordinate the normal way, whenever they run. - A producer can write both at once — tee each record to the live in-PG readers and to a durable materialized output for out-of-PG readers. So fan-out to a mix of co-scheduled and later consumers is expressible; it's not an "impossible" case. - Why this is consistency-clean (not "different failure semantics"): an out-of-PG consumer of the materialized edge is sequenced after group commit (§5 group-observable completion; §7 materialize-before-read). So during a group's internal replay — a sibling fails, the group reruns, the in-PG live readers re-absorb — the materialized consumer hasn't started yet and cannot observe an intermediate attempt. When the group commits, it reads exactly the single committed version, identical to any regular shuffle downstream. There's no window where the producer re-delivers to the transient readers but not the materialized one. The mechanism that guarantees this: a member's materialized output is written as it runs but its map-output registration is deferred to group commit, so a failed attempt's blocks are simply orphaned (never registered) — exactly like a resubmitted regular map stage today. I would like to keep 1:1 mapping for now and defer 1:N later but like I mentioned it is definitely possible to extend it. > 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. Agreed, it's worth stating precisely because the scheduler genuinely permits reuse. The root reason it must be prevented here is simple: a pipelined shuffle is transient — once-through, nothing retained — so unlike a regular shuffle there's no durable output for a second job to reuse. Reuse across jobs is therefore unsound for it, so we make run-once an enforced invariant rather than an assumption. I'll add a spec subsection making this explicit. > This enforcement sounds good to me. Note - there might be similar interaction with checkpoint as well, that would need to be addressed. checkpoint is cache-adjacent but splits into two cases, and I'll handle them differently: - Local checkpoint (localCheckpoint()) is cache — it forces a non-NONE storage level and stores in the block manager (just also truncates lineage). Ephemeral, whole-partition, nothing durable escapes — same safe bucket as .cache()/.persist(), not rejected. - Reliable checkpoint (checkpoint()) is genuinely different, and rather than leave it as a subtle "allowed but hazardous" case, v1 will fail-fast on it — cleaner and less confusing. It writes a durable, lineage-truncated snapshot, which (1) reintroduces exactly the cross-time reuse a transient edge forbids (§4), and (2) is written by a separate post-success job that recomputes the RDD from scratch, which for a PG member means re-reading its already-vanished transient input. I will document it. > 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 ? We could but can we defer it to v2? Initial callers such as RTM would not need such feature. I will mention it explicitly in the spec > In other words, cleanup is triggered by result stage completion, and tied to commit -> 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 ! The distinction I'm drawing is that a PG is an atomic scheduling unit — there are no per-task or per-partition replays within it. That matters because OutputCommitCoordinator is built on the opposite assumption: it arbitrates commits per task attempt, authorizing exactly one attempt per partition and denying any later request for a partition that has already committed. That's sound today because a committed partition is never recomputed — a stage rerun (e.g. on fetch failure) only recomputes the missing partitions, so a task that already committed never needs to commit again, and the permanent "deny" is exactly right. A PG breaks that assumption. Because the group is atomic, a failure anywhere reruns the whole group — including a result stage whose tasks already succeeded and committed. So committed partitions do get rerun, and those rerun tasks must be allowed to commit again — but the coordinator still holds the previous attempt's authorized committers and would deny them (success clears nothing; only a failed holder's slot is cleared today). So to let the rerun's tasks commit, we either (a) rerun the members under fresh stage ids — a fresh coordinator StageState with no prior committers — or (b) reset the committed state for those stages in the OutputCommitCoordinator (e.g. stageEnd on teardown) before the rerun. -- 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]
