Кэш объектов для исполнителей Spark

Хороший вопрос для экспертов Spark.

Я обрабатываю данные в операции с map (RDD). Внутри функции mapper мне нужно искать объекты класса A которые будут использоваться при обработке элементов в RDD.

Поскольку это будет выполняться над исполнителями, а создание элементов типа A (которое будет проверяться), оказывается дорогостоящей операцией, я хочу предварительно загрузить и кэшировать эти объекты для каждого исполнителя. Каков наилучший способ сделать это?

Одна из идей заключается в передаче таблицы поиска, но класс A не является сериализуемым (без контроля над его реализацией).
Другая идея - загрузить их в одноэлементный объект. Тем не менее, я хочу контролировать, что загружается в эту таблицу поиска (например, возможно, разные данные по различным заданиям Spark).

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

Есть ли чистый и элегантный способ сделать это, или его невозможно достичь?

Ответ 1

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

Сериализация часто возникает как проблема при использовании других интерфейсов. Если вы можете обеспечить, чтобы объекты, которые вы потребляете, сериализуемы, это будет лучшим решением. Если это невозможно, ваша жизнь становится немного сложнее. Если вы не можете сериализовать объекты A, вам необходимо создать их для исполнителей для каждой задачи. Если они где-то хранятся в файле, это выглядит примерно так:

rdd.mapPartitions { it => 
  val lookupTable = loadLookupTable(path)
  it.map(elem => fn(lookupTable, elem))
}

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

EDIT: Здесь еще одна модель, которая, я считаю, позволяет вам разделить таблицу поиска по задачам на JVM.

class BroadcastableLookupTable {
  @transient val lookupTable: LookupTable[A] = null

  def get: LookupTable[A] = {
    if (lookupTable == null)
      lookupTable = < load lookup table from disk>
    lookupTable
  }
}

Этот класс может транслироваться (ничего существенного не передается), и при первом вызове на JVM вы загрузите таблицу поиска и вернете ее.

Ответ 2

В случае, если сериализация оказывается невозможной, как насчет сохранения объектов поиска в базе данных? Это не самое простое решение, предоставленное, но должно работать нормально. Я мог бы рекомендовать проверить, например, искра-redis, но я уверен, что там есть лучшее решение.

Ответ 3

Поскольку A не является сериализуемым, самым простым решением является создание собственного сериализуемого типа A1 со всеми данными из A необходимых для вычисления. Затем используйте новую таблицу поиска в трансляции.