数据帧中的并行排列（pandas或dask）

Question

我需要在一个数据帧中按列计算所有可能的行差异排列。你知道吗

使用itertools置换是可行的，但对于大小问题，我需要解决的时间太长了。使用多重处理时出错。假设误差有一个解决方案，“多处理”是一个最佳的方法还是dask有一个解决规模问题的方法？你知道吗

#My naive approach

import pandas as pd
import numpy as np
from itertools import permutations

columns = list(range(1,50))
index = list(range(1,10))
df = pd.DataFrame(index= index, columns = columns,data=np.random.randn(len(index),len(columns)))
count_perm = list(permutations(df.index,2))

comparison_df = pd.DataFrame(columns = df.columns)

for a,b in permutations(df.index,2):
    comparison_df.loc['({} {})'.format(a,b)] = df.loc[a] - df.loc[b]   

#My multiprocessing attempt

import pandas as pd
import numpy as np
from itertools import permutations
from multiprocessing.dummy import Pool as ThreadPool

columns = list(range(1,5000))
index = list(range(1,100))
df = pd.DataFrame(index= index, columns = columns,data=np.random.randn(len(index),len(columns)))
count_perm = list(permutations(df.index,2))

pool = ThreadPool(4)  # Number of threads

comparison_df = pd.DataFrame(columns = df.columns)
aux_val = [(a, b) for a,b in permutations(df.index,2)]

def op(tupx):
    comparison_df.loc["('{}', '{}')".format(tupx[0],tupx[1])]  = (df.loc[tupx[0]] - df.loc[tupx[1]])

pool.map(op, aux_val)

错误：

Traceback (most recent call last):

  File "<ipython-input-69-20c917ebefd7>", line 30, in <module>
    pool.map(op, aux_val)

  File "/home/justaguy/anaconda3/lib/python3.7/multiprocessing/pool.py", line 268, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()

  File "/home/justaguy/anaconda3/lib/python3.7/multiprocessing/pool.py", line 657, in get
    raise self._value

  File "/home/justaguy/anaconda3/lib/python3.7/multiprocessing/pool.py", line 121, in worker
    result = (True, func(*args, **kwds))

  File "/home/justaguy/anaconda3/lib/python3.7/multiprocessing/pool.py", line 44, in mapstar
    return list(map(*args))

  File "<ipython-input-69-20c917ebefd7>", line 26, in op
    comparison_df.loc["('{}', '{}')".format(tupx[0],tupx[1])]  = (df.loc[tupx[0]] - df.loc[tupx[1]])

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/indexing.py", line 190, in __setitem__
    self._setitem_with_indexer(indexer, value)

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/indexing.py", line 451, in _setitem_with_indexer
    self.obj._data = self.obj.append(value)._data

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/frame.py", line 6692, in append
    sort=sort)

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/reshape/concat.py", line 229, in concat
    return op.get_result()

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/reshape/concat.py", line 426, in get_result
    copy=self.copy)

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/internals/managers.py", line 2065, in concatenate_block_managers
    return BlockManager(blocks, axes)

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/internals/managers.py", line 114, in __init__
    self._verify_integrity()

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/internals/managers.py", line 311, in _verify_integrity
    construction_error(tot_items, block.shape[1:], self.axes)

  File "/home/justaguy/anaconda3/lib/python3.7/site-packages/pandas/core/internals/managers.py", line 1691, in construction_error
    passed, implied))

ValueError: Shape of passed values is (604, 4999), indices imply (602, 4999)

Answer 1

正如我在评论中建议的那样，您可能会认为使用combinations而不是permutations。这样做可以减少一半的计算量。免责声明：我的代码计算的是列之间的差异，而不是您的示例中的索引。你知道吗

import pandas as pd
import numpy as np
from itertools import permutations, combinations
import os 
import multiprocessing as mp

# generate data
columns = list(range(1,50))

## I don't think you should start index at 1
index = list(range(1,10))

df = pd.DataFrame(index=index,
                  columns=columns,
                  data=np.random.randn(len(index),len(columns)))

单线程

%%timeit -n 10
df1 = pd.DataFrame()
for a,b in permutations(df.index,2):
    df1["{}-{}".format(a,b)] = df[a]-df[b]
# 37.1 ms ± 726 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

%%timeit -n 10
df1 = pd.DataFrame()
for a,b in permutations(df.index,2):
    df1["{}-{}".format(a,b)] = df[a].values-df[b].values

df1.index = df1.index+1
# 25.6 ms ± 1.2 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)

单线程-使用组合

%%timeit -n 10
df1 = pd.DataFrame()
for a,b in combinations(df.index,2):
    df1["{}-{}".format(a,b)] = df[a]-df[b]
# 18.6 ms ± 1.07 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)

%%timeit -n 10
df1 = pd.DataFrame()
for a,b in combinations(df.index,2):
    df1["{}-{}".format(a,b)] = df[a].values-df[b].values

df1.index = df1.index+1
# 13.2 ms ± 819 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

多重处理

在这种情况下，这不会更快，但您可以考虑将其用于其他应用程序。你知道吗

def parallelize(fun, vec, cores):
    with mp.Pool(cores) as p:
        res = p.map(fun, vec)
    return res

def fun(v):
    a,b=v
    cols = ["{}-{}".format(a,b)]
    df_out = pd.DataFrame(data=df[a].values-df[b].values,
                          columns=cols)

    return df_out

vec = [(a,b) for a,b in permutations(df.index,2)]
cores = os.cpu_count()

%%timeit -n 10
df1 = parallelize(fun, vec, cores)
df1 = pd.concat(df1, axis=1)
# 260 ms ± 10.7 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)