В искры Apache, в чем разница между использованием mapPartitions и объединением использования широковещательной переменной и карты

Question

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

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

Так в чем же разница между этими двумя? Можем ли мы использовать это interchangebly только для общих переменных?

Answer 1

broadcast предназначен для отправки объекта каждому рабочему узлу. Этот объект будет разделяться между всеми разделами на этом узле (и объект value/i.e будет одинаковым для каждого узла в кластере). Цель вещания - экономить на сетевых расходах, когда вы используете одни и те же данные во многих разных задачах/разделах на рабочем узле.

mapPartitions, наоборот, является методом, доступным на RDD, и работает как map, только на разделах. Да, вы можете определить новые объекты, такие как соединение jdbc, которые затем будут уникальными для каждого раздела. Однако вы не можете делиться им между разными разделами и тем более между разными узлами.

Answer 2

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

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

object DummySharedState { 
    var i = 0L 
    def get(x: Any) = { 
    i += 1L 
    i 
    } 
} 

sc.parallelize(1 to 100, 1).map(DummySharedState.get).max 
// res3: Long = 100 
sc.parallelize(1 to 100, 2).map(DummySharedState.get).max 
// res4: Long = 50 
sc.parallelize(1 to 100, 50).map(DummySharedState.get).max 
// res5: Long = 2 

и подобная вещь в PySpark:

• модуль одноточечно dummy_shared_state.py:

  i = 0 
  def get(x): 
      global i 
      i += 1 
      return i 
  

• Основной сценарий:

  from pyspark import SparkConf, SparkContext 
  import dummy_shared_state 
  
  master = "spark://..." 
  conf = (SparkConf() 
      .setMaster(master) 
      .set("spark.python.worker.reuse", "false")) 
  
  sc.addPyFile("dummy_shared_state.py") 
  sc.parallelize(range(100), 1).map(dummy_shared_state.get).max() 
  ## 100 
  sc.parallelize(range(100), 2).map(dummy_shared_state.get).max() 
  ## 50 
  

Обратите внимание, что для параметра spark.python.worker.reuse установлено значение false.Если оставить значение по умолчанию, вы на самом деле увидеть что-то вроде этого:

sc.parallelize(range(100), 2).map(dummy_shared_state.get).max() 
## 50 
sc.parallelize(range(100), 2).map(dummy_shared_state.get).max() 
## 100 
sc.parallelize(range(100), 2).map(dummy_shared_state.get).max() 
## 150 

В конце дня вы должны различать три разных вещи:

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

Кроме того, существуют некоторые специфические ошибки Python, связанные с использованием постоянных интерпретаторов.

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