Dask 连接导致内存错误，当熊猫连接不用于相同的文件时

Question

我正在尝试连接几个 dask 数据帧，但这会导致我的所有 RAM 都被用完并导致我的环境 (Google Colab) 崩溃。

我曾尝试与 Dask 连接，因为我听说 Dask 对文件进行分区，以便更轻松地加载到内存中。然而，Pandas 能够处理他的操作，而 Dask 则不能。

我使用 Dask 的原因是因为当我尝试保存我的 Pandas 数据框时，我的环境崩溃了。所以我想看看 Dask 是否能够在不崩溃的情况下保存我的数据，但我一直在创建我的数据框。

combA = np.load(file2A.format(0) , allow_pickle=True)
combB = np.load(file2B.format(0), allow_pickle=True )
combC = np.load(file2C.format(0), allow_pickle=True )
combD = np.load(file2D.format(0) , allow_pickle=True)
combE = np.load(file2E.format(0) , allow_pickle=True )
combF = np.load(file2F.format(0), allow_pickle=True )

dfAllA = dd.from_pandas(pd.DataFrame(combA), npartitions=10)
dfAllB = dd.from_pandas(pd.DataFrame(combB), npartitions=10)
dfAllC = dd.from_pandas(pd.DataFrame(combC), npartitions=10)
dfAllD = dd.from_pandas(pd.DataFrame(combD), npartitions=10)
dfAllE = dd.from_pandas(pd.DataFrame(combE), npartitions=10)
dfAllF = dd.from_pandas(pd.DataFrame(combF), npartitions=10)

dfAllT = dd.concat([dfAllA, dfAllB, dfAllC, dfAllD, dfAllE, dfAllF ], interleave_partitions=True)

我想在没有内存错误的情况下执行连接。

从下面的答案看来，我应该定义一个函数来执行日期的加载和连接，将其输入到 dask.delayed 函数中，然后对这些函数执行 .compute()

类似

def daskFunc1():

    combA = np.load(file2A.format(0) , allow_pickle=True)
    combB = np.load(file2B.format(0), allow_pickle=True )
    combC = np.load(file2C.format(0), allow_pickle=True )
    combD = np.load(file2D.format(0) , allow_pickle=True)
    combE = np.load(file2E.format(0) , allow_pickle=True )
    combF = np.load(file2F.format(0), allow_pickle=True )

    dfAllA = dd.from_pandas(pd.DataFrame(combA), npartitions=10)
    dfAllB = dd.from_pandas(pd.DataFrame(combB), npartitions=10)
    dfAllC = dd.from_pandas(pd.DataFrame(combC), npartitions=10)
    dfAllD = dd.from_pandas(pd.DataFrame(combD), npartitions=10)
    dfAllE = dd.from_pandas(pd.DataFrame(combE), npartitions=10)
    dfAllF = dd.from_pandas(pd.DataFrame(combF), npartitions=10)

def daskFunc2():
    dfAllT = dd.concat([dfAllA, dfAllB, dfAllC, dfAllD, dfAllE, dfAllF ], interleave_partitions=True)

from dask.delayed import delayed

f1 = delayed(daskFunc1)
f2 = delayed(daskFunc2)

f1.compute()
f2.compute()

但是，当我尝试时

dfAllT.head()

我明白了

NameError: name 'dfAllT' is not defined

Answer 1

您当前正在将所有数据加载到 RAM 中，然后将其交给 Dask。如果你的所有数据在第一次启动时已经填满了 RAM，Dask 就帮不了你了。

相反，告诉 Dask 如何加载数据并让它在正确的时间加载会更好。本文档可能会为您指明正确的方向：https://docs.dask.org/en/latest/delayed-collections.html，这是一个旧示例 https://gist.github.com/mrocklin/e7b7b3a65f2835cda813096332ec73ca