Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/7276#discussion_r35225996
--- Diff:
streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
---
@@ -354,46 +503,22 @@ class ReceiverTracker(ssc: StreamingContext,
skipReceiverLaunch: Boolean = false
rcvr
})
- val checkpointDirOption = Option(ssc.checkpointDir)
- val serializableHadoopConf =
- new SerializableConfiguration(ssc.sparkContext.hadoopConfiguration)
-
- // Function to start the receiver on the worker node
- val startReceiver = (iterator: Iterator[Receiver[_]]) => {
- if (!iterator.hasNext) {
- throw new SparkException(
- "Could not start receiver as object not found.")
- }
- val receiver = iterator.next()
- val supervisor = new ReceiverSupervisorImpl(
- receiver, SparkEnv.get, serializableHadoopConf.value,
checkpointDirOption)
- supervisor.start()
- supervisor.awaitTermination()
- }
-
// Run the dummy Spark job to ensure that all slaves have registered.
// This avoids all the receivers to be scheduled on the same node.
if (!ssc.sparkContext.isLocal) {
ssc.sparkContext.makeRDD(1 to 50, 50).map(x => (x,
1)).reduceByKey(_ + _, 20).collect()
}
- // Get the list of executors and schedule receivers
- val executors = getExecutors(ssc)
- val tempRDD =
- if (!executors.isEmpty) {
- val locations = scheduleReceivers(receivers, executors)
- val roundRobinReceivers = (0 until receivers.length).map(i =>
- (receivers(i), locations(i)))
- ssc.sc.makeRDD[Receiver[_]](roundRobinReceivers)
- } else {
- ssc.sc.makeRDD(receivers, receivers.size)
- }
-
// Distribute the receivers and start them
logInfo("Starting " + receivers.length + " receivers")
running = true
+
try {
- ssc.sparkContext.runJob(tempRDD,
ssc.sparkContext.clean(startReceiver))
+ for (receiver <- receivers) {
+ endpoint.send(StartReceiver(receiver))
+ }
+ // Wait until all receivers exit
+ receiverJobExitLatch.await()
logInfo("All of the receivers have been terminated")
} finally {
running = false
--- End diff --
Since we don't need to block, the thread is not necessary. Then we can also
remove `ReceiverLauncher`. Therefore, I removed `ReceiverLauncher` and added a
method `launchReceivers` to replace it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
