dask.async.MemoryError 而运行 EC2 上的大数据计算

Question

我有一个 m4.4xlarge（64 GB 内存）EC2 盒子。我正在运行与 pandas 约会。我收到以下内存错误。

我在运行大约 24 小时后收到此消息，这大约是完成任务所需的时间，所以我不确定错误是否是由于 RAM、磁盘内存不足造成的脚本的末尾我执行 DF.to_csv() 将大型 DF 写入磁盘或 pandas/numpy 内部内存限制？

raise(remote_exception(res, tb))
    dask.async.MemoryError: 

Traceback
---------
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/dask/async.py", line 267, in execute_task
    result = _execute_task(task, data)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/dask/async.py", line 248, in _execute_task
    args2 = [_execute_task(a, cache) for a in args]
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/dask/async.py", line 249, in _execute_task
    return func(*args2)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/frame.py", line 4061, in apply
    return self._apply_standard(f, axis, reduce=reduce)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/frame.py", line 4179, in _apply_standard
    result = result._convert(datetime=True, timedelta=True, copy=False)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/generic.py", line 3004, in _convert
    copy=copy)).__finalize__(self)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 2941, in convert
    return self.apply('convert', **kwargs)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 2901, in apply
    bm._consolidate_inplace()
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 3278, in _consolidate_inplace
    self.blocks = tuple(_consolidate(self.blocks))
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 4269, in _consolidate
    _can_consolidate=_can_consolidate)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 4289, in _merge_blocks
    new_values = _vstack([b.values for b in blocks], dtype)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 4335, in _vstack
    return np.vstack(to_stack)
  File "/home/ec2-user/anaconda2/lib/python2.7/site-packages/numpy/core/shape_base.py", line 230, in vstack
    return _nx.concatenate([atleast_2d(_m) for _m in tup], 0)

更新：

所以根据 MRocklin 的回答提供了一些额外的信息。

以下是我执行流程的方式：

def dask_stats_calc(dfpath,v1,v2,v3...):
    dfpath_ddf = dd.from_pandas(dfpath,npartitions=16,sort=False)
    return dfpath_ddf.apply(calculate_stats,axis=1,args=(dfdaily,v1,v2,v3...)).compute(get=get).stack().reset_index(drop=True)

f_threaded = partial(dask_stats_calc,dfpath,v1,v2,v3...,multiprocessing.get)
f_threaded()

现在 dfpath 是一个有 140 万行的 df，因此 dfpath_ddf.apply() 运行超过 140 万行。

一旦整个 dfpath_ddf.apply() 完成，就会发生 df.to_csv()，但正如您所说，最好定期写入磁盘。

现在的问题是，我如何实现类似于每隔 20 万行定期写入磁盘的操作？我想我可以将 dfpath_ddf 分解成 200k 块（或类似的东西）并且每个顺序运行？

Answer 1

单线程执行

有时，在等待写入磁盘上的单个文件 时，任务会在 RAM 中堆积。使用像这样的顺序输出对于并行系统来说本质上是棘手的。如果您需要使用单个文件，那么我建议您尝试使用相同的单线程计算，看看是否有所不同。

with dask.set_options(get=dask.async.get_sync):
    DF.to_csv('out.csv')

写入多个文件

或者（最好）您可以尝试写入许多 CSV 文件。这在调度上要容易得多，因为任务不必等到它们的前任完成才能写入磁盘并从 RAM 中删除它们自己。

DF.to_csv('out.*.csv')

例子

因此，一种常见且相当稳健的并行执行和写入方式是将您的计算与最后对 to_csv 的调用结合起来

ddf = dd.from_pandas(df, npartitions=100)
ddf.apply(myfunc).to_csv('out.*.csv')

这会将您的数据帧分解成块，在每个块上调用您的函数，将该块写入磁盘，然后删除中间值，释放 space。

dask.async.MemoryError 而运行 EC2 上的大数据计算

dask.async.MemoryError while running big data computation on EC2

python

memory

pandas

dask

单线程执行

写入多个文件

例子

dask.async.MemoryError 而 运行 EC2 上的大数据计算

dask.async.MemoryError while running big data computation on EC2

python

memory

pandas

dask

单线程执行

写入多个文件

例子

dask.async.MemoryError 而运行 EC2 上的大数据计算