python - pandas 数据帧上的多重处理。基于输入大小的令人困惑的行为

Question

我正在尝试实现类似 df.apply 的函数，但在数据帧的 block 之间并行化。我编写了以下测试代码来看看我可以获得多少 yield (与数据复制等相比):

from multiprocessing import Pool
from functools import partial
import pandas as pd
import numpy as np
import time

def df_apply(df, f):
    return df.apply(f, axis=1)

def apply_in_parallel(df, f, n=5):
    pool = Pool(n)
    df_chunks = np.array_split(df, n)
    apply_f = partial(df_apply, f=f)
    result_list = pool.map(apply_f, df_chunks)
    return pd.concat(result_list, axis=0)

def f(x):
  return x+1

if __name__ == '__main__':
  N = 10^8
  df = pd.DataFrame({"a": np.zeros(N), "b": np.zeros(N)})

  print "parallel"
  t0 = time.time()
  r = apply_in_parallel(df, f, n=5)
  print time.time() - t0

  print "single"
  t0 = time.time()
  r = df.apply(f, axis=1)
  print time.time() - t0

奇怪的行为:对于 N=10^7 有效对于 N=10^8 它给了我一个错误

Traceback (most recent call last):
  File "parallel_apply.py", line 27, in <module>
    r = apply_in_parallel(df, f, n=5)
  File "parallel_apply.py", line 14, in apply_in_parallel
    result_list = pool.map(apply_f, df_chunks)
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 227, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 528, in get
    raise self._value
AttributeError: 'numpy.ndarray' object has no attribute 'apply'

有人知道这是怎么回事吗？我也非常感谢有关这种并行化方式的任何反馈。我期望函数比每行和数百万行的 inc 或 sum 花费更多的时间。

谢谢!

Answer 1

array_split 接受任何类似数组的参数(包括 pandas.DataFrame 对象)，但仅返回保证它返回一个 numpy.ndarray (DataFrame不是)。当然，ndarrays 没有 apply 方法，这正是您所看到的错误。事实上，我很惊讶这在任何情况下都有效。您需要将数据帧拆分为子帧或应用在 ndarray 上运行的函数。