Функция уменьшения искры: понять, как это работает

Question

Я принимаю это course.

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

Я думал, что функция начнет работать на обеих машинах в то же время, а затем, когда он имеет результаты от 2-х машин, он будет снова запустить функцию в последний раз

rdd1=rdd.reduce(lambda x,y: x+y) 

обновление 1-- ------------------------------------------

будет ниже шагов дать более быстрый ответ по сравнению с функцией уменьшения?

Rdd=[3,5,4,7,4] 
seqOp = (lambda x, y: x+y) 
combOp = (lambda x, y: x+y) 
collData.aggregate(0, seqOp, combOp) 

Update 2 -----------------------------------

Если оба набор кодов ниже выполняется за такое же количество времени? Я проверил, и кажется, что и то и другое.

import datetime 

data=range(1,1000000000) 
distData = sc.parallelize(data,4) 
print(datetime.datetime.now()) 
a=distData.reduce(lambda x,y:x+y) 
print(a) 
print(datetime.datetime.now()) 

seqOp = (lambda x, y: x+y) 
combOp = (lambda x, y: x+y) 
print(datetime.datetime.now()) 
b=distData.aggregate(0, seqOp, combOp) 
print(b) 
print(datetime.datetime.now()) 

Answer 1

reduce поведение немного отличается между нативной (Scala) и языков гостей (Python), но упрощает вещи немного:

• каждый раздел обрабатывается последовательно поэлементно
• несколько разделов может обрабатываться одновременно одним работником (несколько потоков исполнителей) или разными работниками
• Частичные результаты отбираются водителю, где применяется заключительная редукция (это часть, которая имеет di fferent внедрение в PySpark и Scala)

Так это выглядит, как вы используете Python позволяет взглянуть на код:

1. reducecreates a simple wrapper для пользователя при условии функции:

   def func(iterator): 
       ... 
   

2. Это обертка is used to mapPartitions:

   vals = self.mapPartitions(func).collect() 
   

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

3. Собранный vals уменьшается последовательно на водителя, используя стандартный Python reduce:

   reduce(f, vals) 
   

   где f является a функции, переданные по адресу RDD.reduce

Для сравнения Scala will mer ge частичные результаты асинхронно, поскольку они исходят от рабочих.

В случае treeReduce шаг 3. может выполняться также распределенным образом. См Understanding treeReduce() in Spark

reduce Резюмируя, за исключением обработки со стороны водителя, использует одни и те же механизмы (mapPartitions) в качестве основных преобразований, как map или filter и обеспечивают тот же уровень параллелизма (раз за исключением кода драйвера). Если у вас большое количество разделов или f стоит дорого, вы можете параллелизировать/распространять окончательное слияние с использованием tree* семейства методов.