Problema

Estou disposto a fazer uma engenharia de recursos usando o módulo de multiprocessamento(multiprocessing.Pool.starmap(). No entanto, ele fornece uma mensagem de erro da seguinte maneira. Eu acho que essa mensagem de erro é do tamanho de entradas (2147483647 = 2 ^ 31 - 1?), Pois o mesmo código funcionou sem problemas por uma fração(frac=0.05) de quadros de dados de entrada (train_scala, test, ts). Eu converto os tipos de quadro de dados o menor possível, no entanto, não melhora.

A versão anaconda é 4.3.30 e a versão Python é 3.6 (64 bits). E o tamanho da memória do sistema é superior a 128 GB, com mais de 20 núcleos. Gostaria de sugerir algum ponteiro ou solução para superar esse problema? Se esse problema for causado por dados grandes para um módulo de multiprocessamento, quantos dados menores devo usar para utilizar o módulo de multiprocessamento no Python3?

Código:

from multiprocessing import Pool, cpu_count
from itertools import repeat    
p = Pool(8)
is_train_seq = [True]*len(historyCutoffs)+[False]
config_zip = zip(historyCutoffs, repeat(train_scala), repeat(test), repeat(ts), ul_parts_path, repeat(members), is_train_seq)
p.starmap(multiprocess_FE, config_zip)

Mensagem de erro:

Traceback (most recent call last):
  File "main_1210_FE_scala_multiprocessing.py", line 705, in <module>
    print('----Pool starmap start----')
  File "/home/dmlab/ksedm1/anaconda3/envs/py36/lib/python3.6/multiprocessing/pool.py", line 274, in starmap
    return self._map_async(func, iterable, starmapstar, chunksize).get()
  File "/home/dmlab/ksedm1/anaconda3/envs/py36/lib/python3.6/multiprocessing/pool.py", line 644, in get
    raise self._value
  File "/home/dmlab/ksedm1/anaconda3/envs/py36/lib/python3.6/multiprocessing/pool.py", line 424, in _handle_tasks
    put(task)
  File "/home/dmlab/ksedm1/anaconda3/envs/py36/lib/python3.6/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/home/dmlab/ksedm1/anaconda3/envs/py36/lib/python3.6/multiprocessing/connection.py", line 393, in _send_bytes
    header = struct.pack("!i", n)
struct.error: 'i' format requires -2147483648 <= number <= 2147483647

Informações adicionaishistoryCutoffs é uma lista de números inteirostrain_scala é um DataFrame de pandas (377MB)teste é um DataFrame do pandas (15MB)ts é um DataFrame de pandas (547MB)ul_parts_path é uma lista de diretórios (string)is_train_seq é uma lista de booleanos

Código extra: método multiprocess_FE

def multiprocess_FE(historyCutoff, train_scala, test, ts, ul_part_path, members, is_train):
    train_dict = {}
    ts_dict = {}
    msno_dict = {}
    ul_dict = {}
    if is_train == True:
        train_dict[historyCutoff] = train_scala[train_scala.historyCutoff == historyCutoff]
    else:
        train_dict[historyCutoff] = test
    msno_dict[historyCutoff] = set(train_dict[historyCutoff].msno)
    print('length of msno is {:d} in cutoff {:d}'.format(len(msno_dict[historyCutoff]), historyCutoff))
    ts_dict[historyCutoff] = ts[(ts.transaction_date <= historyCutoff) & (ts.msno.isin(msno_dict[historyCutoff]))]
    print('length of transaction is {:d} in cutoff {:d}'.format(len(ts_dict[historyCutoff]), historyCutoff))    
    ul_part = pd.read_csv(gzip.open(ul_part_path, mode="rt"))  ##.sample(frac=0.01, replace=False)
    ul_dict[historyCutoff] = ul_part[ul_part.msno.isin(msno_dict[historyCutoff])]
    train_dict[historyCutoff] = enrich_by_features(historyCutoff, train_dict[historyCutoff], ts_dict[historyCutoff], ul_dict[historyCutoff], members, is_train)