Почему мне нужно использовать Broadcast для того, чтобы общий доступ с rdd уже транслировался?

Question

введите описание изображения здесь В DAGScheduler's submitMissingTasks(....) есть код taskBinary = sc.broadcast(taskBinaryBytes), СДР с функцией уже переданы.

А затем ShuffleMapTask и ResultTask десериализуют TaskBinary для запуска на Исполнителе.Таким образом, искра помогает нам отправлять переменные общего доступа исполнителю при выполнении широковещательных заданий.

Почему мне нужно использовать sc.broadcast(xxxxxx) самостоятельно?

Я понял это. Раньше я не знал механизма десериализации.Хотя Spark aleardy транслирует RDD и совместно использует переменные, но если мы не используем широковещательную рассылку, когда задачи десериализации получат разные переменные общего доступа (десериализация много раз, десериализация задачи не может аннулироваться.), Это займет больше времени и больше памяти.

В нашем коде широковещание только помогает сократить время и память на этапе десериализации, а не сериализовать и скопировать задачу на этап исполнителя.

Answer 1

Потому что, если вы просто используете локальную переменную, т.е. val data = list(1, 4, 5), Spark упакует это и отправит вместе с вашим кодом для каждой задачи!

В случае трансляции val broadcastData = sc.broadcast(data) будет скопирован на каждого исполнителя и, следовательно, доступен локально.

В некоторых случаях у вас могут быть данные большого размера, и вы хотите избежать сериализации этих данных для каждой задачи, вместо этого вы можете отправить их каждому исполнителю с помощью broadcast, тогда эти задачи могут быть доступны локально для этих задач.

UPDATE:

Как упомянул @Fish, кажется, что Spark внутренне использует широковещание для распространения данных задач, это можно увидеть в методе submitMissingTasks здесь:

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

Хотя сериализованный размер задачи не должен превышать 1 МБ, как определено TaskSetManager.TASK_SIZE_TO_WARN_KIB.