Почему этот код Scala, очевидно, не работает на рабочих Spark, а только на узле драйвера Spark?

Question

Я использовал упомянутый код здесь для создания HashMap в Scala. Ниже для удобства вставьте копию:

def genList(xx: String) = {
    Seq("one", "two", "three", "four")
  }

  val oriwords = Set("hello", "how", "are", "you")

  val newMap = (Map[String, (String, Int)]() /: oriwords) (
    (cmap, currentWord) => {
      val xv = 2

      genList(currentWord).foldLeft(cmap) {
        (acc, ps) => {
          val src = acc get ps

          if (src == None) {
            acc + (ps -> ((currentWord, xv)))
          }
          else {
            if (src.get._2 < xv) {
              acc + (ps -> ((currentWord, xv)))
            }
            else acc
          }

        }
      }
    }
  )

  println(newMap)

Примечание : приведенный выше код работает для малых значений oriwords, однако он не работает, если значение oriwords велико. Очевидно, потому что вычисления происходят на узле драйвера Spark.

При запуске я получаю исключение «Недостаточно памяти» следующим образом:

 WARN  HeartbeatReceiver:66 - Removing executor driver with no recent heartbeats: 159099 ms exceeds timeout 120000 ms
Exception in thread "dispatcher-event-loop-1"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "dispatcher-event-loop-1"
Exception in thread "refresh progress" java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space

Как заставить вычисление выполняться в кластере Spark и сохранить сгенерированный HashMap в самом кластере Spark вместо его вычисления и сохранения в узле драйвера Spark?

Answer 1

Spark использует абстракции DataFrame и RDD для представления данных. Он не использует Scala Maps. Поэтому вам нужно обернуть ваши данные в RDD или DataFrame (предпочтительный вариант). В зависимости от типа данных, которые у вас есть, существуют разные способы загрузки данных.

Answer 2

Вещи должны быть в RDD, Dataset, Dataframe и др. для искры, чтобы распространять ваши вычисления. в основном все происходит в драйвере, за исключением вещей, которые есть в HoF, таких как map и foreach в одной из этих структур.