Python: использование многопроцессорной обработки на pandas фрейме данных

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

import pandas as pd
from geopy.distance import vincenty
from itertools import combinations
import multiprocessing as mp

df = pd.DataFrame({'ser_no': [1, 2, 3, 4, 5, 6, 7, 8, 9, 0],
                'co_nm': ['aa', 'aa', 'aa', 'bb', 'bb', 'bb', 'bb', 'cc', 'cc', 'cc'],
                'lat': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
                'lon': [21, 22, 23, 24, 25, 26, 27, 28, 29, 30]})



def calc_dist(x):
    return pd.DataFrame(
               [ [grp,
                  df.loc[c[0]].ser_no,
                  df.loc[c[1]].ser_no,
                  vincenty(df.loc[c[0], x], 
                           df.loc[c[1], x])
                 ]
                 for grp,lst in df.groupby('co_nm').groups.items()
                 for c in combinations(lst, 2)
               ],
               columns=['co_nm','machineA','machineB','distance'])

if __name__ == '__main__':
    pool = mp.Pool(processes = (mp.cpu_count() - 1))
    pool.map(calc_dist, ['lat','lon'])
    pool.close()
    pool.join()

Я использую Python 2.7.11 и Ipython 4.1.2 с Anaconda 2.5.0 64-бит в Windows7 Professional при возникновении этой ошибки.

runfile ('C:/.../Desktop/multiprocessing test.py', wdir = 'C:/.../Desktop') Traceback (последний последний звонок):

Файл ", строка 1, в     runfile ('C:/.../Desktop/multiprocessing test.py', wdir = 'C:/.../Desktop')

Файл" C:...\Local\Continuum\Anaconda2\lib\site-packages\spyderlib\widgets\externalshell\sitecustomize.py ", строка 699, в файле запуска     execfile (имя файла, пространство имен)

Файл" C:...\Local\Continuum\Anaconda2\lib\site-packages\spyderlib\widgets\externalshell\sitecustomize.py ", строка 74, в execfile     exec (компиляция (scripttext, filename, 'exec'), glob, loc)

Файл" C:/..../multiprocessing test.py ", строка 33, в     pool.map(calc_dist, ['lat', 'lon'])

Файл" C:...\AppData\Local\Continuum\Anaconda2\lib\multiprocessing\pool.py ", строка 251, на карте     return self.map_async (func, iterable, chunksize).get()

Файл" C:...\Local\Continuum\Anaconda2\lib\multiprocessing\pool.py", строка 567, в get     raise self._value

TypeError: не удалось создать экземпляр Point из 1.

def get(self, timeout=None):
    self.wait(timeout)
    if not self._ready:
        raise TimeoutError
    if self._success:
        return self._value
    else:
        raise self._value

Ответ 1

Что случилось

Эта строка из вашего кода:

pool.map(calc_dist, ['lat','lon'])

порождает 2 процесса - один запускает calc_dist('lat'), а другой запускает calc_dist('lon'). Сравните первый пример в doc. (В принципе, pool.map(f, [1,2,3]) вызывает f три раза с аргументами, приведенными в следующем списке: f(1), f(2) и f(3).) Если я не ошибаюсь, ваша функция calc_dist может быть только называется calc_dist('lat', 'lon'). И это не допускает параллельной обработки.

Решение

Я считаю, что вы хотите разделить работу между процессами, возможно, отправив каждый кортеж (grp, lst) в отдельный процесс. Следующий код делает именно это.

Сначала давайте подготовиться к расщеплению:

grp_lst_args = list(df.groupby('co_nm').groups.items())

print(grp_lst_args)
[('aa', [0, 1, 2]), ('cc', [7, 8, 9]), ('bb', [3, 4, 5, 6])]

Мы отправим каждый из этих кортежей (здесь их три) в качестве аргумента функции в отдельном процессе. Нам нужно переписать функцию, назовем ее calc_dist2. Для удобства этот аргумент является кортежем, как в calc_dist2(('aa',[0,1,2]))

def calc_dist2(arg):
    grp, lst = arg
    return pd.DataFrame(
               [ [grp,
                  df.loc[c[0]].ser_no,
                  df.loc[c[1]].ser_no,
                  vincenty(df.loc[c[0], ['lat','lon']], 
                           df.loc[c[1], ['lat','lon']])
                 ]
                 for c in combinations(lst, 2)
               ],
               columns=['co_nm','machineA','machineB','distance'])

И теперь идет многопроцессорная обработка:

pool = mp.Pool(processes = (mp.cpu_count() - 1))
results = pool.map(calc_dist2, grp_lst_args)
pool.close()
pool.join()

results_df = pd.concat(results)

results - это список результатов (здесь кадры данных) вызовов calc_dist2((grp,lst)) для (grp,lst) в grp_lst_args. Элементы results позже объединяются в один кадр данных.

print(results_df)
  co_nm  machineA  machineB          distance
0    aa         1         2  156.876149391 km
1    aa         1         3  313.705445447 km
2    aa         2         3  156.829329105 km
0    cc         8         9  156.060165391 km
1    cc         8         0  311.910998169 km
2    cc         9         0  155.851498134 km
0    bb         4         5  156.665641837 km
1    bb         4         6  313.214333025 km
2    bb         4         7  469.622535339 km
3    bb         5         6  156.548897414 km
4    bb         5         7  312.957597466 km
5    bb         6         7   156.40899677 km

Кстати, в Python 3 мы могли бы использовать конструкцию with:

with mp.Pool() as pool:
    results = pool.map(calc_dist2, grp_lst_args)

Обновление

Я тестировал этот код только на linux. В linux доступ к кадру данных только для чтения df может выполняться дочерними процессами и не копируется в их пространство памяти, но я не уверен, как это работает в Windows. Вы можете рассмотреть разбиение df на куски (сгруппированные по co_nm) и отправка этих кусков в качестве аргументов в другую версию calc_dist.

Ответ 2

Странно. Кажется, он работает под python2, но не python3.

Это минимально модифицированная версия для печати результата:

import pandas as pd
from geopy.distance import vincenty
from itertools import combinations
import multiprocessing as mp

df = pd.DataFrame({'ser_no': [1, 2, 3, 4, 5, 6, 7, 8, 9, 0],
                'co_nm': ['aa', 'aa', 'aa', 'bb', 'bb', 'bb', 'bb', 'cc', 'cc', 'cc'],
                'lat': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
                'lon': [21, 22, 23, 24, 25, 26, 27, 28, 29, 30]})



def calc_dist(x):
    ret =  pd.DataFrame(
               [ [grp,
                  df.loc[c[0]].ser_no,
                  df.loc[c[1]].ser_no,
                  vincenty(df.loc[c[0], x],
                           df.loc[c[1], x])
                 ]
                 for grp,lst in df.groupby('co_nm').groups.items()
                 for c in combinations(lst, 2)
               ],
               columns=['co_nm','machineA','machineB','distance'])
    print(ret)
    return ret

if __name__ == '__main__':
    pool = mp.Pool(processes = (mp.cpu_count() - 1))
    pool.map(calc_dist, ['lat','lon'])
    pool.close()
    pool.join()

И это результат работы python2

0     aa         1         2  110.723608682 km
1     aa         1         3  221.460709525 km
2     aa         2         3  110.737100843 km
3     cc         8         9  110.827576495 km
4     cc         8         0  221.671650552 km
   co_nm  machineA  machineB          distance
5     cc         9         0  110.844074057 km
0     aa         1         2  110.575064814 km
1     aa         1         3  221.151481337 km
6     bb         4         5  110.765515243 km
2     aa         2         3  110.576416524 km
7     bb         4         6    221.5459187 km
3     cc         8         9  110.598565514 km
4     cc         8         0  221.203121352 km
8     bb         4         7  332.341640771 km
5     cc         9         0  110.604555838 km
6     bb         4         5   110.58113908 km
9     bb         5         6  110.780403457 km
7     bb         4         6  221.165643396 km
10    bb         5         7  221.576125528 km
8     bb         4         7  331.754177186 km
9     bb         5         6  110.584504316 km
10    bb         5         7  221.173038106 km
11    bb         6         7  110.795722071 km
11    bb         6         7   110.58853379 km

И это трассировка стека из python3

"""
Traceback (most recent call last):
  File "/usr/local/lib/python3.4/dist-packages/geopy/point.py", line 123, in __new__
    seq = iter(arg)
TypeError: 'numpy.int64' object is not iterable

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.4/multiprocessing/pool.py", line 119, in worker
    result = (True, func(*args, **kwds))
  File "/usr/lib/python3.4/multiprocessing/pool.py", line 44, in mapstar
    return list(map(*args))
  File "gps.py", line 29, in calc_dist
    for grp, lst in df.groupby('co_nm').groups.items()
  File "gps.py", line 30, in <listcomp>
    for c in combinations(lst, 2)
  File "/usr/local/lib/python3.4/dist-packages/geopy/distance.py", line 322, in __init__
    super(vincenty, self).__init__(*args, **kwargs)
  File "/usr/local/lib/python3.4/dist-packages/geopy/distance.py", line 115, in __init__
    kilometers += self.measure(a, b)
  File "/usr/local/lib/python3.4/dist-packages/geopy/distance.py", line 342, in measure
    a, b = Point(a), Point(b)
  File "/usr/local/lib/python3.4/dist-packages/geopy/point.py", line 126, in __new__
    "Failed to create Point instance from %r." % (arg,)
TypeError: Failed to create Point instance from 8.
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "gps.py", line 38, in <module>
    pool.map(calc_dist, ['lat', 'lon'])
  File "/usr/lib/python3.4/multiprocessing/pool.py", line 260, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/usr/lib/python3.4/multiprocessing/pool.py", line 599, in get
    raise self._value
TypeError: Failed to create Point instance from 8.

Я знаю, что это не ответ, но, возможно, это помогает...