为啥Pandas的DataFrame使用Series而非直接使用矩阵进行存储和数据操纵?
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了为啥Pandas的DataFrame使用Series而非直接使用矩阵进行存储和数据操纵?相关的知识,希望对你有一定的参考价值。
参考技术APandas的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
计算,它会超级快(比这个并行化快很多)。
您可以使用multiprocessing
或joblib
来实现并行化。但是,如果组的数量很大并且每个组的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 defined
:combinedDf = pd.concat(parallel_list)
?
伊万,很明显!我认为他的答案非常好。 hackity hack 比我的少得多。以上是关于为啥Pandas的DataFrame使用Series而非直接使用矩阵进行存储和数据操纵?的主要内容,如果未能解决你的问题,请参考以下文章