Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/21385#discussion_r190354487
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/shuffle/UnsafeRowReceiver.scala
---
@@ -41,11 +50,15 @@ private[shuffle] case class ReceiverEpochMarker()
extends UnsafeRowReceiverMessa
*/
private[shuffle] class UnsafeRowReceiver(
queueSize: Int,
+ numShuffleWriters: Int,
+ checkpointIntervalMs: Long,
override val rpcEnv: RpcEnv)
extends ThreadSafeRpcEndpoint with ContinuousShuffleReader with
Logging {
// Note that this queue will be drained from the main task thread and
populated in the RPC
// response thread.
- private val queue = new
ArrayBlockingQueue[UnsafeRowReceiverMessage](queueSize)
+ private val queues = Array.fill(numShuffleWriters) {
--- End diff --
I agree. This level of issues can be dealt with later once we run this and
observed the issues for real. Also, it's likely that we will eventually replace
a whole lot of this with a non-RPC-endpoint-based transfer mechanism. So lets
keep something very simple for now.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
