viirya commented on code in PR #58097:
URL: https://github.com/apache/spark/pull/58097#discussion_r3817078964
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ShuffleExchangeExec.scala:
##########
@@ -570,7 +570,12 @@ object ShuffleExchangeExec {
rddWithPartitionIds,
new PartitionIdPassthrough(part.numPartitions),
serializer,
- shuffleWriterProcessor = createShuffleWriteProcessor(writeMetrics),
+ // Copy rows only for a transport that hands object references to a
concurrent
+ // consumer (the in-process channel). The RPC streaming transport
detaches rows by
+ // serializing them promptly and must not pay an extra per-row copy
on its path.
+ shuffleWriterProcessor = createShuffleWriteProcessor(
+ writeMetrics,
+ copyRows =
SparkEnv.get.pipelinedShuffleManager.requiresDetachedRecords),
Review Comment:
Both fixed. (1) `copyRows = requiresDetachedRecords &&
!needToCopyObjectsBeforeShuffle(part)`, so the write processor no longer copies
a row that `prepareShuffleDependency` already copied. (2)
`ChannelShuffleWriter` now stores the incoming pair directly instead of
re-wrapping it in a fresh `Tuple2` -- one fewer allocation per record.
##########
core/src/main/scala/org/apache/spark/shuffle/local/pipelined/ChannelShuffleRendezvous.scala:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.local.pipelined
+
+import java.util.concurrent.{ConcurrentHashMap, LinkedBlockingQueue}
+
+/**
+ * Process-wide rendezvous between the map (writer) and reduce (reader) sides
of an
+ * in-process pipelined shuffle. One bounded queue exists per
+ * `(shuffleId, reducePartitionId)`; every map task writing to a given reduce
partition
+ * shares the queue with the single reduce task that drains it. Queue elements
are BATCHES
+ * of records (`Array[AnyRef]` of pairs, see [[ChannelShuffleWriter]]) or the
+ * [[EndOfStream]] marker, so the queue's per-operation lock cost is paid per
batch, not
+ * per row.
+ *
+ * This is correct only when producer and consumer tasks are co-resident in
the same JVM,
+ * i.e. a single executor (local mode). The concurrent-stage scheduler
co-schedules the two
+ * stages so both are running, but it does not by itself guarantee
co-location; the
+ * [[PipelinedChannelShuffleManager]] is only intended for single-executor
deployments,
+ * where co-location is automatic. Cross-executor pipelined shuffle is served
by the RPC
+ * streaming shuffle instead.
+ *
+ * The queues are bounded, so a fast producer blocks on `put` when the
consumer lags --
+ * this is the backpressure that keeps the pipelined hand-off memory-bounded.
+ */
+private[spark] object ChannelShuffleRendezvous {
+
+ /**
+ * Marker placed on a queue by each map task when it finishes writing to
that reduce
+ * partition. A reader stops once it has seen one marker per map task.
+ */
+ val EndOfStream: AnyRef = new AnyRef
+
+ // Keyed by (shuffleId, reducePartitionId). Values are AnyRef because the
queue carries
+ // both record batches (Array[AnyRef]) and the EndOfStream marker.
+ private val queues =
+ new ConcurrentHashMap[(Int, Int), LinkedBlockingQueue[AnyRef]]()
+
+ // (shuffleId, reducePartitionId) keys whose reader has departed (its reduce
task finished)
+ // and will drain no more. A writer stops feeding an abandoned partition and
drops the rest.
+ // This covers the LIVE-partition early-stop case (e.g. a LIMIT reader that
pulled enough
+ // and quit): without it the writer fills the partition's bounded queue and
blocks forever.
+ private val abandoned =
+ java.util.concurrent.ConcurrentHashMap.newKeySet[(Int, Int)]()
+
+ /**
+ * Default per-queue capacity in BATCHES (not rows): with the default
1024-row batch this
+ * bounds the in-flight hand-off at ~64K rows per reduce partition.
+ */
+ private val DefaultCapacity = 64
Review Comment:
Fixed by promoting it to a conf:
`spark.shuffle.pipelined.channel.queueCapacity` (default 64), and its doc
states the `queueCapacity * batchSize * numPartitions` heap-residency product
so a user tuning either knob can reason about it. The channel manager sets the
rendezvous capacity from it at construction.
--
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]
