Dask：为什么内存使用量会激增？答案

【问题标题】：Dask: why is memory usage blowing up?Dask：为什么内存使用量会激增？
【发布时间】：2021-06-22 17:56:31
【问题描述】：

我有一个小数据帧（约 100MB）和一个昂贵的计算，我想为每一行执行。它不是可向量化的计算；它需要对每一行进行一些解析和数据库查找。

因此，我决定尝试使用 Dask 来并行化任务。该任务是“令人尴尬的并行”，执行顺序或重复执行没有问题。但是，由于某些未知原因，内存使用量激增至约 100GB。

这是有问题的代码示例：

import pandas as pd
import numpy as np
import dask.dataframe as dd

from dask.distributed import Client
from dask_jobqueue import LSFCluster

cluster = LSFCluster(memory="6GB", cores=1, project='gRNA Library Design')
cluster.scale(jobs=16)
client = Client(cluster)

required_dict = load_big_dict()
score_guide = lambda row: expensive_computation(required_dict, row)

library_df = pd.read_csv(args.library_csv)

meta = library_df.dtypes
meta = meta.append(pd.Series({
    'specificity': np.dtype('int64'),
    'cutting_efficiency': np.dtype('int64'), 
    '0 Off-targets': np.dtype('object'),
    '1 Off-targets': np.dtype('object'),
    '2 Off-targets': np.dtype('object'),
    '3 Off-targets': np.dtype('object')}))
    
library_ddf = dd.from_pandas(library_df, npartitions=32)
library_ddf = library_ddf.apply(score_guide, axis=1, meta=meta)
library_ddf = library_ddf.compute()
library_ddf = library_ddf.drop_duplicates()
library_ddf.to_csv(args.outfile, index=False)

我的猜测是，查找所需的大字典不知何故是问题所在，但它的大小总共只有约 1.5GB，并且不包含在生成的数据帧中。

为什么 Dask 会占用内存？

【问题讨论】：

标签： python pandas data-science dask

【解决方案1】：

不是 100% 确定这会在这种情况下解决它，但您可以尝试 futurize 字典：

# broadcasting makes sure that every worker has a copy
[fut_dict] = client.scatter([required_dict], broadcast=True)
score_guide = lambda row: expensive_computation(fut_dict, row)

这样做的目的是将字典的副本放在每个工作人员上，并将对对象的引用存储在 fut_dict 中，从而避免每次调用函数时都需要 hash 大字典：

每次你传递一个具体的结果（任何没有延迟的东西）时，Dask 都会默认对它进行哈希处理，给它一个名字。这是相当快的（大约 500 MB/s），但如果你一遍又一遍地这样做可能会很慢。相反，最好也延迟您的数据。

请注意，这会占用每个工作人员的一部分内存（例如，根据您的信息，每个工作人员将有 1.5GB 的空间分配给 dict）。您可以在Q&A 中阅读更多内容。

【讨论】：

我尝试了与此类似的操作，但未能使其正常工作（可能是我的错误）。您已经正确识别了问题：将大字典作为全局变量。由于截止日期，我实施了一个更简单的解决方案。我将更新我的解决方案，并在明天调查其他人。感谢您的帮助。
几周前我遇到了类似的问题。不知何故，使用数据作为全局变量会重载调度程序工作人员而不是单个工作人员。例如在某些时候，我在类的构造函数中加载数据，这是一个非常糟糕的主意 :) 你应该朝那个方向看

【解决方案2】：

问题是required_dict 需要被序列化并发送到所有工作线程。由于required_dict 很大，并且许多工作人员同时需要它，因此重复序列化会导致大量内存爆炸。

有很多修复；对我来说，最简单的方法是从工作线程中加载字典并明确使用map_partitions 而不是apply。

这是代码中的解决方案，

    def do_df(df):
        required_dict = load_big_dict()
        score_guide = lambda row: expensive_computation(required_dict, row)
        return df.apply(score_guide, axis=1)
        
    library_ddf = dd.from_pandas(library_df, npartitions=128)
    library_ddf = library_ddf.map_partitions(do_df)
    library_ddf = library_ddf.compute()

【讨论】：