python struct.error: 'i' 格式需要 -2147483648 <= number <= 2147483647

Question

问题

我愿意使用多处理模块(multiprocessing.Pool.starmap() 进行特征工程。 但是，它会给出如下错误消息。我猜这个错误消息与输入的大小有关（2147483647 = 2^31 - 1？），因为相同的代码对于输入数据帧的分数(frac=0.05)（train_scala，test，ts）运行顺利。我将数据帧的类型转换为尽可能小，但它并没有变得更好。

anaconda 版本是 4.3.30，Python 版本是 3.6（64 位）。 系统内存超过128GB，20多核。 您想提出任何解决此问题的建议或解决方案吗？如果这个问题是由于多处理模块的大数据引起的，我应该使用多少小数据来利用 Python3 上的多处理模块？

代码：

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)

错误信息：

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

额外信息

• historyCutoffs 是一个整数列表
• train_scala 是一个 pandas DataFrame (377MB)
• test 是一个 pandas DataFrame (15MB)
• ts 是 pandas DataFrame (547MB)
• ul_parts_path 是目录列表（字符串）
• is_train_seq 是一个布尔值列表

额外代码：方法 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)

Answer 1

进程间的通信协议使用pickling，pickled数据以pickled数据的大小为前缀。对于您的方法，所有参数一起被腌制为一个对象。

您生成的对象在腌制后大于i 结构格式化程序（一个四字节有符号整数）的大小，这打破了代码所做的假设。

您可以将数据帧的读取委托给子进程，只发送加载数据帧所需的元数据。它们的总大小接近 1GB，在您的进程之间通过管道共享的数据太多了。

引用Programming guidelines section:

比pickle/unpickle更好继承

当使用spawn 或forkserver 启动方法时，multiprocessing 中的许多类型需要是可挑选的，以便子进程可以使用它们。 但是，通常应该避免使用管道或队列将共享对象发送到其他进程。相反，您应该安排程序，以便需要访问在其他地方创建的共享资源的进程可以从祖先进程继承它。

如果您没有在 Windows 上运行并使用 spawn 或 forkserver 方法，您可以在启动子进程之前将数据帧加载为全局变量，此时子进程将“通过正常的 OS 写入时复制内存页面共享机制继承数据。

请注意，在 Python 3.8 中，非 Windows 系统的此限制已提高到 unsigned long long（8 字节），因此您现在可以发送和接收 4EiB 的数据。请参阅 this commit，以及 Python 问题 #35152 和 #17560。

如果你不能升级，你不能利用资源继承，并且不在Windows上运行，那么使用这个补丁：

import functools
import logging
import struct
import sys

logger = logging.getLogger()


def patch_mp_connection_bpo_17560():
    """Apply PR-10305 / bpo-17560 connection send/receive max size update

    See the original issue at https://bugs.python.org/issue17560 and 
    https://github.com/python/cpython/pull/10305 for the pull request.

    This only supports Python versions 3.3 - 3.7, this function
    does nothing for Python versions outside of that range.

    """
    patchname = "Multiprocessing connection patch for bpo-17560"
    if not (3, 3) < sys.version_info < (3, 8):
        logger.info(
            patchname + " not applied, not an applicable Python version: %s",
            sys.version
        )
        return

    from multiprocessing.connection import Connection

    orig_send_bytes = Connection._send_bytes
    orig_recv_bytes = Connection._recv_bytes
    if (
        orig_send_bytes.__code__.co_filename == __file__
        and orig_recv_bytes.__code__.co_filename == __file__
    ):
        logger.info(patchname + " already applied, skipping")
        return

    @functools.wraps(orig_send_bytes)
    def send_bytes(self, buf):
        n = len(buf)
        if n > 0x7fffffff:
            pre_header = struct.pack("!i", -1)
            header = struct.pack("!Q", n)
            self._send(pre_header)
            self._send(header)
            self._send(buf)
        else:
            orig_send_bytes(self, buf)

    @functools.wraps(orig_recv_bytes)
    def recv_bytes(self, maxsize=None):
        buf = self._recv(4)
        size, = struct.unpack("!i", buf.getvalue())
        if size == -1:
            buf = self._recv(8)
            size, = struct.unpack("!Q", buf.getvalue())
        if maxsize is not None and size > maxsize:
            return None
        return self._recv(size)

    Connection._send_bytes = send_bytes
    Connection._recv_bytes = recv_bytes

    logger.info(patchname + " applied")

Answer 2

此问题已在最近对 python 的 PR 中修复 https://github.com/python/cpython/pull/10305

如果您愿意，您可以在本地进行此更改，使其立即为您工作，而无需等待 python 和 anaconda 发布。