使用dask.bag和pandas.DataFrame将字典的dask.delayed转换为dask.dataframe

Question

我正在努力将字典的dask.bag转换为dask.delayed pandas.DataFrames，使之成为最终的dask.dataframe

我有一个函数(make_dict)将文件读入相当复杂的嵌套字典结构，另一个函数(make_df)将这些字典转换为pandas.DataFrame (由此产生的数据为每个文件大约100 mb )。为了进一步分析，我想将所有数据文件附加到一个dask.dataframe中。

到目前为止，我一直在使用dask.delayed对象来加载、转换和附加所有工作正常的数据(参见下面的示例)。但是，对于以后的工作，我想使用dask.bag将加载的字典存储在dask.persist()中。

我成功地将数据加载到dask.bag中，生成了一个在调用compute()之后可以在本地使用的dicts列表或pandas.DataFrame列表。然而，当我尝试使用dask.bag将dask.dataframe转换为dask.dataframe时，我遇到了一个错误(参见下面)。

在这里，我感觉好像遗漏了一些相当简单的东西，或者我的dask.bag方法是错误的？

下面的示例展示了我使用简化函数的方法，并抛出了相同的错误。任何关于如何解决这一问题的建议都是值得赞赏的。

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

print(dask.__version__) # 1.1.4
print(pd.__version__) # 0.24.2

def make_dict(n=1):
    return {"name":"dictionary","data":{'A':np.arange(n),'B':np.arange(n)}}

def make_df(d):
    return pd.DataFrame(d['data'])

k = [1,2,3]

# using dask.delayed
dfs = []
for n in k:
    delayed_1 = dask.delayed(make_dict)(n)
    delayed_2 = dask.delayed(make_df)(delayed_1)
    dfs.append(delayed_2)
ddf1 = dask.dataframe.from_delayed(dfs).compute() # this works as expected

# using dask.bag and turning bag of dicts into bag of DataFrames
b1 = dask.bag.from_sequence(k).map(make_dict)
b2 = b1.map(make_df)

df = pd.DataFrame().append(b2.compute()) # <- I would like to do this using delayed dask.DataFrames like above
ddf2 = dask.dataframe.from_delayed(b2.to_delayed()).compute() # <- this fails

# error:
# ValueError: Expected iterable of tuples of (name, dtype), got [   A  B
# 0  0  0]

我最终想要使用分布式调度程序做的事情：

b = dask.bag.from_sequence(k).map(make_dict)
b = b.persist()
ddf = dask.dataframe.from_delayed(b.map(make_df).to_delayed())

Answer 1

在这种情况下，延迟的对象指向元素列表，所以您有一个熊猫数据列表，这并不完全是您想要的。两项建议

1. 只要坚持使用dask.delayed。对你来说似乎很好
2. 使用数据帧方法，它需要一袋数据集，并自己进行数据转换。