为啥Pandas的DataFrame使用Series而非直接使用矩阵进行存储和数据操纵?

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了为啥Pandas的DataFrame使用Series而非直接使用矩阵进行存储和数据操纵?相关的知识,希望对你有一定的参考价值。

参考技术A

Pandas的DataFrame使用Series而非直接使用矩阵进行存储和数据操纵的原因是因为Series可以为每一列或行提供标签(label)。这使得我们可以通过标签来查询和操作数据,而不是通过矩阵的位置索引。这样做可以更加方便和直观地处理数据,特别是当数据集较大时。同时DataFrame使用了多个Series来存储数据,可以使得DataFrame具有了数据关联性,这对数据分析是非常有帮助的。

在 pandas groupby 之后并行化应用

【中文标题】在 pandas groupby 之后并行化应用【英文标题】:Parallelize apply after pandas groupby 【发布时间】:2014-11-29 00:46:19 【问题描述】:

我在groupby之后使用rosetta.parallel.pandas_easy并行化apply,例如:

from rosetta.parallel.pandas_easy import groupby_to_series_to_frame
df = pd.DataFrame('a': [6, 2, 2], 'b': [4, 5, 6],index= ['g1', 'g1', 'g2'])
groupby_to_series_to_frame(df, np.mean, n_jobs=8, use_apply=True, by=df.index)

但是,有没有人想出如何并行化返回 DataFrame 的函数?正如预期的那样,此代码对于 rosetta 失败。

def tmpFunc(df):
    df['c'] = df.a + df.b
    return df

df.groupby(df.index).apply(tmpFunc)
groupby_to_series_to_frame(df, tmpFunc, n_jobs=1, use_apply=True, by=df.index)

【问题讨论】:

【参考方案1】:

编辑:为了在 pandas groupby 上获得更好的计算性能,您可以使用 numba 在运行时将您的代码编译成 C 代码并以 C 速度运行。如果你在groupby 之后应用的函数是纯numpy 计算,它会超级快(比这个并行化快很多)。

您可以使用multiprocessingjoblib 来实现并行化。但是,如果组的数量很大并且每个组的DataFrame很大,则运行时间可能会更糟,因为您需要将这些组多次转移到CPU中。为了减少开销,我们可以先将数据分成大块,然后在这些块上并行计算。

例如,假设您正在处理股票数据,您需要按代码对股票进行分组,然后计算一些统计数据。您可以先按代码的第一个字符(大块)进行分组,然后在这个虚拟组中执行操作:

import pandas as pd
from joblib import Parallel, delayed

def group_func(dummy_group):
    # Do something to the group just like doing to the original dataframe.
    #     Example: calculate daily return.
    res = []
    for _, g in dummy_group.groupby('code'):
        g['daily_return']  = g.close / g.close.shift(1)
        res.append(g)
    return pd.concat(res)

stock_data = stock_data.assign(dummy=stock_data['code'].str[0])

Parallel(n_jobs=-1)(delayed(group_func)(group) for _, group in stock_data.groupby('dummy'))

【讨论】:

【参考方案2】:

Ivan 的回答很棒,不过看起来可以稍微简化一下,也不需要依赖joblib:

from multiprocessing import Pool, cpu_count

def applyParallel(dfGrouped, func):
    with Pool(cpu_count()) as p:
        ret_list = p.map(func, [group for name, group in dfGrouped])
    return pandas.concat(ret_list)

顺便说一句:这不能代替 any groupby.apply(),但它会涵盖典型情况:例如它应该涵盖案例 2 和案例 3 in the documentation,而您应该通过将参数 axis=1 提供给最终的 pandas.concat() 调用来获得案例 1 的行为。

编辑: 文档已更改;旧版本可以在here找到,无论如何我复制粘贴下面的三个例子。

case 1: group DataFrame apply aggregation function (f(chunk) -> Series) yield DataFrame, with group axis having group labels

case 2: group DataFrame apply transform function ((f(chunk) -> DataFrame with same indexes) yield DataFrame with resulting chunks glued together

case 3: group Series apply function with f(chunk) -> DataFrame yield DataFrame with result of chunks glued together

【讨论】:

@Keiku 不知道,我以前从未听说过 REPL...但是您是否尝试过 func = lambda x : x"? If this doesn't work either, I suggest you open a specific question. You should be able to reproduce just with applyParallel([('one', 1), ('two', 2)] , your_func)`` 感谢您的建议。看来我尝试重新启动控制台并解决了它。很抱歉给您添麻烦了。 文档似乎不再提供示例。有人可以详细说明一下吗?【参考方案3】:

根据this thread,我个人建议使用 dask。

正如@chrisb 所指出的,在 python 中使用 pandas 进行多处理可能会产生不必要的开销。它也可能像多线程甚至单线程一样执行。

Dask 是专门为多进程创建的。

【讨论】:

【参考方案4】:

伴随 JD Long 的回答的简短评论。我发现如果组的数量非常大(比如数十万),并且您的应用功能正在做一些相当简单和快速的事情,那么将您的数据帧分成块并将每个块分配给一个工作人员以执行groupby-apply(串行)比并行 groupby-apply 和让工作人员从包含多个组的队列中读取要快得多。示例:

import pandas as pd
import numpy as np
import time
from concurrent.futures import ProcessPoolExecutor, as_completed

nrows = 15000
np.random.seed(1980)
df = pd.DataFrame('a': np.random.permutation(np.arange(nrows)))

所以我们的数据框看起来像:

    a
0   3425
1   1016
2   8141
3   9263
4   8018

请注意,“a”列有很多组(想想客户 ID):

len(df.a.unique())
15000

对我们的组进行操作的函数:

def f1(group):
    time.sleep(0.0001)
    return group

启动一个池:

ppe = ProcessPoolExecutor(12)
futures = []
results = []

做一个并行的groupby-apply:

%%time

for name, group in df.groupby('a'):
    p = ppe.submit(f1, group)
    futures.append(p)

for future in as_completed(futures):
    r = future.result()
    results.append(r)

df_output = pd.concat(results)
del ppe

CPU times: user 18.8 s, sys: 2.15 s, total: 21 s
Wall time: 17.9 s

现在让我们添加一个列,将 df 划分为更少的组:

df['b'] = np.random.randint(0, 12, nrows)

现在只有 12 个,而不是 15000 个组:

len(df.b.unique())
12

我们将对 df 进行分区并对每个块执行 groupby-apply。

ppe = ProcessPoolExecutor(12)

包装乐趣:

def f2(df):
    df.groupby('a').apply(f1)
    return df

依次发送每个要操作的块:

%%time

for i in df.b.unique():
    p = ppe.submit(f2, df[df.b==i])
    futures.append(p)

for future in as_completed(futures):
    r = future.result()
    results.append(r)

df_output = pd.concat(results) 

CPU times: user 11.4 s, sys: 176 ms, total: 11.5 s
Wall time: 12.4 s

请注意,每组花费的时间量没有改变。相反,改变的是工作人员从中读取的队列的长度。我怀疑正在发生的事情是工作人员无法同时访问共享内存,并且不断返回以读取队列,从而相互踩踏。使用较大的块进行操作,工人返回的频率较低,因此这个问题得到了改善,整体执行速度更快。

【讨论】:

在我有 4 个物理内核的机器上,如果 f1 的延迟增加,我只能看到并行化的好处,否则串行有更好的时间。【参考方案5】:

这似乎可行,虽然它确实应该内置在 pandas 中

import pandas as pd
from joblib import Parallel, delayed
import multiprocessing

def tmpFunc(df):
    df['c'] = df.a + df.b
    return df

def applyParallel(dfGrouped, func):
    retLst = Parallel(n_jobs=multiprocessing.cpu_count())(delayed(func)(group) for name, group in dfGrouped)
    return pd.concat(retLst)

if __name__ == '__main__':
    df = pd.DataFrame('a': [6, 2, 2], 'b': [4, 5, 6],index= ['g1', 'g1', 'g2'])
    print 'parallel version: '
    print applyParallel(df.groupby(df.index), tmpFunc)

    print 'regular version: '
    print df.groupby(df.index).apply(tmpFunc)

    print 'ideal version (does not work): '
    print df.groupby(df.index).applyParallel(tmpFunc)

【讨论】:

你知道在pandas中加入并行化是否有进展吗? 通过对函数进行小的修改,可以返回常规应用返回的分层索引:def temp_func(func, name, group): return func(group), name def applyParallel(dfGrouped, func): retLst, top_index = zip(*Parallel(n_jobs=multiprocessing.cpu_count())(delayed(temp_func)(func, name, group) for name, group in dfGrouped)) return pd.concat(retLst, keys=top_index) Dang,我不知道如何在 cmets 中发布代码... 您应该能够通过将applyParallel 绑定到df 来使“理想版本”工作:from types import MethodType; df.applyParallel = MethodType(applyParallel, df) 我已经尝试过这种方法,但它并没有使用所有可用的 cpu,它只使用 1 或 2,即使我有 8 - 有人发生过这种情况吗? 小心,这最终会比单核版本慢!如果您向每个作业发送大量数据但只有很短的计算,那么开销不值得,并且最终会变慢。【参考方案6】:

我有一个用于在 Pandas 中实现并行化的 hack。我将我的数据帧分成块,将每个块放入列表的元素中,然后使用 ipython 的并行位对数据帧列表进行并行应用。然后我使用 pandas concat 函数将列表重新组合在一起。

但是,这并不普遍适用。它对我有用,因为我想应用于数据帧的每个块的功能大约需要一分钟。并且将我的数据拆分和组合起来并不需要那么长时间。所以这显然是一个杂牌。话虽如此,这里有一个例子。我正在使用 Ipython 笔记本,所以你会在我的代码中看到 %%time 魔法:

## make some example data
import pandas as pd

np.random.seed(1)
n=10000
df = pd.DataFrame('mygroup' : np.random.randint(1000, size=n), 
                   'data' : np.random.rand(n))
grouped = df.groupby('mygroup')

对于本例,我将根据上述 groupby 制作“块”,但这不一定是数据的分块方式。虽然这是一个很常见的模式。

dflist = []
for name, group in grouped:
    dflist.append(group)

设置并行位

from IPython.parallel import Client
rc = Client()
lview = rc.load_balanced_view()
lview.block = True

编写一个愚蠢的函数来应用我们的数据

def myFunc(inDf):
    inDf['newCol'] = inDf.data ** 10
    return inDf

现在让我们先串行然后并行运行代码。 串行优先:

%%time
serial_list = map(myFunc, dflist)
CPU times: user 14 s, sys: 19.9 ms, total: 14 s
Wall time: 14 s

现在并行

%%time
parallel_list = lview.map(myFunc, dflist)

CPU times: user 1.46 s, sys: 86.9 ms, total: 1.54 s
Wall time: 1.56 s

那么只需几毫秒就可以将它们合并回一个数据帧

%%time
combinedDf = pd.concat(parallel_list)
 CPU times: user 296 ms, sys: 5.27 ms, total: 301 ms
Wall time: 300 ms

我在我的 MacBook 上运行了 6 个 IPython 引擎,但您可以看到它将执行时间从 14 秒降至 2 秒。

对于真正长时间运行的随机模拟,我可以通过使用 StarCluster 启动集群来使用 AWS 后端。然而,很多时候,我只在我的 MBP 上跨 8 个 CPU 进行并行化。

【讨论】:

我会用我的代码试试这个,谢谢。你能解释一下为什么 apply 不会自动并行化操作吗?似乎拥有 apply 函数的全部好处是避免循环,但如果它不对这些组执行此操作,有什么好处? 由于 GIL,Python 中的并行化很困难的说法由来已久。请记住,apply 通常是语法糖,并且在它下面执行隐含循环。使用并行化有点棘手,因为并行化存在运行时成本,这有时会抵消并行化的好处。 parallel_list 是否缺少定义,因为它在此行出现错误name 'parallel_list' is not definedcombinedDf = pd.concat(parallel_list) 伊万,很明显!我认为他的答案非常好。 hackity hack 比我的少得多。

以上是关于为啥Pandas的DataFrame使用Series而非直接使用矩阵进行存储和数据操纵?的主要内容,如果未能解决你的问题,请参考以下文章

pandas

Pandas数据分析

pandas层次化索引

为啥Pandas的DataFrame使用Series而非直接使用矩阵进行存储和数据操纵?

在 pandas groupby 之后并行化应用

为啥在具有一级索引的 MultiIndex 列的 pandas DataFrame 中表现不同?