提问人：jf328 提问时间：7/12/2015 最后编辑：gnatjf328 更新时间：3/17/2019 访问量：27226
python dask DataFrame，支持（微不足道的并行化）行应用？
python dask DataFrame, support for (trivially parallelizable) row apply?
问：
我最近发现了 dask 模块，它旨在成为一个易于使用的 python 并行处理模块。对我来说，最大的卖点是它适用于熊猫。
在阅读了其手册页面后，我找不到一种方法来完成这个微不足道的并行化任务：
ts.apply(func) # for pandas series
df.apply(func, axis = 1) # for pandas DF row apply
目前，为了在 dask 中实现这一目标，AFAIK，
ddf.assign(A=lambda df: df.apply(func, axis=1)).compute() # dask DataFrame
这是丑陋的语法，实际上比直接慢
df.apply(func, axis = 1) # for pandas DF row apply
有什么建议吗？
编辑：感谢@MRocklin的地图功能。它似乎比普通的熊猫应用要慢。这是否与熊猫GIL发布问题有关，还是我做错了？
import dask.dataframe as dd
s = pd.Series([10000]*120)
ds = dd.from_pandas(s, npartitions = 3)
def slow_func(k):
A = np.random.normal(size = k) # k = 10000
s = 0
for a in A:
if a > 0:
s += 1
else:
s -= 1
return s
s.apply(slow_func) # 0.43 sec
ds.map(slow_func).compute() # 2.04 sec
答：
74赞
MRocklin
7/12/2015
#1
map_partitions
您可以使用该函数将函数应用于数据帧的所有分区。map_partitions
df.map_partitions(func, columns=...)
请注意，func 一次只会提供数据集的一部分，而不是像 with 那样提供整个数据集（如果您想进行并行性，您可能不希望这样做）。pandas apply
map / apply
您可以使用以下命令在系列中逐行映射函数map
df.mycolumn.map(func)
您可以使用以下命令在数据帧中按行映射函数apply
df.apply(func, axis=1)
线程与进程
从版本 0.6.0 开始，与线程并行化。自定义 Python 函数不会从基于线程的并行性中获得太多好处。您可以尝试流程dask.dataframes
df = dd.read_csv(...)
df.map_partitions(func, columns=...).compute(scheduler='processes')
但要避免apply
但是，您确实应该避免在 Pandas 和 Dask 中使用自定义 Python 函数。这通常是性能不佳的根源。如果您找到一种以矢量化方式进行操作的方法，那么您的 Pandas 代码可能会快 100 倍，并且您根本不需要 dask.dataframe。apply
考虑numba
对于您的特定问题，您可以考虑 numba。这可以显著提高您的性能。
In [1]: import numpy as np
In [2]: import pandas as pd
In [3]: s = pd.Series([10000]*120)
In [4]: %paste
def slow_func(k):
A = np.random.normal(size = k) # k = 10000
s = 0
for a in A:
if a > 0:
s += 1
else:
s -= 1
return s
## -- End pasted text --
In [5]: %time _ = s.apply(slow_func)
CPU times: user 345 ms, sys: 3.28 ms, total: 348 ms
Wall time: 347 ms
In [6]: import numba
In [7]: fast_func = numba.jit(slow_func)
In [8]: %time _ = s.apply(fast_func) # First time incurs compilation overhead
CPU times: user 179 ms, sys: 0 ns, total: 179 ms
Wall time: 175 ms
In [9]: %time _ = s.apply(fast_func) # Subsequent times are all gain
CPU times: user 68.8 ms, sys: 27 µs, total: 68.8 ms
Wall time: 68.7 ms
免责声明，我为制造和雇用许多开发人员的公司工作。numbadaskpandas
评论
0赞
jf328
7/12/2015
谢谢！我尝试了地图方法，它似乎比熊猫应用慢。你能对原始帖子的编辑发表评论吗？
0赞
jf328
7/12/2015
我已经经常使用numba了！谢谢你的工作。我通常做的是，DataFrame 的每一行都指定模拟的配置（复杂/慢速函数的参数）。我已经在做多处理了，只是要注意更好的方法
0赞
Bob Haffner
7/13/2015
@MRocklin 关于熊猫有点跑题了;我尝试使用地图而不是应用，因为我听说它更快，但我不确定为什么它更快。如能提供任何澄清或澄清链接，将不胜感激。
2赞
MRocklin
7/13/2015
@BobHaffner毫无头绪。建议做一个小实验，并发布一个“为什么会这样”风格的堆栈溢出问题。
2赞
Shubham Chaudhary
6/30/2017
#2
截至 v .apply 责任：dask.dataframemap_partitions
@insert_meta_param_description(pad=12)
def apply(self, func, convert_dtype=True, meta=no_default, args=(), **kwds):
""" Parallel version of pandas.Series.apply
...
"""
if meta is no_default:
msg = ("`meta` is not specified, inferred from partial data. "
"Please provide `meta` if the result is unexpected.\n"
" Before: .apply(func)\n"
" After: .apply(func, meta={'x': 'f8', 'y': 'f8'}) for dataframe result\n"
" or: .apply(func, meta=('x', 'f8')) for series result")
warnings.warn(msg)
meta = _emulate(M.apply, self._meta_nonempty, func,
convert_dtype=convert_dtype,
args=args, **kwds)
return map_partitions(M.apply, self, func,
convert_dtype, args, meta=meta, **kwds)
评论
daskmultiprocessingnp.array_splitmultiprocessingpd.concat