Как распараллелить цикл for в python / pyspark (который может быть запущен на нескольких узлах на серверах Amazon)? - PullRequest
Новая
       45

Как распараллелить цикл for в python / pyspark (который может быть запущен на нескольких узлах на серверах Amazon)?

0 голосов
/ 02 июня 2019

Извините, если это ужасно простой вопрос, но я просто не могу найти простой ответ на мой запрос.

У меня есть некоторый вычислительно интенсивный код, который смущающе распараллеливается.Псевдокод выглядит следующим образом. 

n = 500
rounds = 200

data = [d_1, ..., d_n]
values = [ 0 for _ in range(n) ]

for _ in range(rounds):
  for i in range(n): # Inner Loop
    values[i] = compute_stuff(data[i])
  data = special_function(values)

Каждая итерация внутреннего цикла занимает 30 секунд, но они полностью независимы.Поэтому я хочу запустить итерации n=500 параллельно, разделив вычисления на 500 отдельных узлов, работающих в Amazon, сократив время выполнения внутреннего цикла до ~ 30 секунд.Как мне это сделать?

Я предполагаю, что PySpark - это стандартная среда, которую можно использовать для этого, а Amazon EMR - это соответствующий сервис, который позволил бы мне выполнять это на многих узлах параллельно.Итак, мой вопрос: как мне дополнить приведенный выше код, чтобы он работал на 500 параллельных узлах на серверах Amazon с использованием инфраструктуры PySpark?Или же, есть ли другая платформа и / или сервис Amazon, которые я должен использовать для этого?

Вот некоторые подробности о псевдокоде.Каждая запись данных d_i является пользовательским объектом, хотя ее можно преобразовать (и восстановить из) в 2 массива чисел A и B, если это необходимо.Возвращаемое значение compute_stuff (и, следовательно, каждая запись values) также является пользовательским объектом.Хотя, опять же, этот пользовательский объект может быть преобразован (и восстановлен из) в словарь списков чисел.Также, compute_stuff требует использования PyTorch и NumPy.Наконец, special_function не такая простая вещь, как сложение, поэтому я не могу использовать ее как «уменьшающую» часть ванильной карты-редукции.

Любая помощь приветствуется!

1 Ответ

1 голос
/ 03 июня 2019

На основании вашего описания я бы не стал использовать pyspark. Чтобы обработать ваши данные с помощью pyspark, вам необходимо полностью переписать код (просто назвать несколько вещей: использование rdd, использование функций spark вместо функций python). Я думаю, что гораздо проще (в вашем случае!) Использовать что-то вроде замечательного pymp . Вам не нужно много менять свой код: 

#still pseudocode
import pymp

n = 500
rounds = 200

data = [d_1, ..., d_n]
values = pymp.shared.list()

for _ in range(rounds):
  with pymp.Parallel(n) as p:
        for i in p.range(n):
            values.append(compute_stuff(data[i]))
  data = special_function(values)

Если важен порядок вашего списка values, вы можете использовать p.thread_num +i для расчета отличительных индексов. Pymp позволяет вам использовать все ядра вашей машины. Если вы хотите использовать несколько aws-машин, вы должны взглянуть на slurm .

...