将 Pandas DataFrame 的 Dask Bag 转换为单个 Dask DataFrame答案

【问题标题】：Convert Dask Bag of Pandas DataFrames to a single Dask DataFrame将 Pandas DataFrame 的 Dask Bag 转换为单个 Dask DataFrame
【发布时间】：2019-12-13 14:32:20
【问题描述】：

问题总结

短版

我如何从一个 Dask Bag of Pandas DataFrame 转换为单个 Dask DataFrame？

加长版

我有一些文件无法被 dask.dataframe 的各种 read 函数（例如 dd.read_csv 或 dd.read_parquet）读取。我确实有自己的函数，可以将它们作为 Pandas DataFrames 读取（函数一次只能处理一个文件，类似于pd.read_csv）。我想将所有这些单个 Pandas DataFrames 放在一个大的 Dask DataFrame 中。

最小工作示例

这里有一些示例 CSV 数据（我的数据实际上不是 CSV 格式，但在此处使用它是为了便于示例）。要创建一个最小的工作示例，您可以将其保存为 CSV 并复制几份，然后使用下面的代码

"gender","race/ethnicity","parental level of education","lunch","test preparation course","math score","reading score","writing score"
"female","group B","bachelor's degree","standard","none","72","72","74"
"female","group C","some college","standard","completed","69","90","88"
"female","group B","master's degree","standard","none","90","95","93"
"male","group A","associate's degree","free/reduced","none","47","57","44"
"male","group C","some college","standard","none","76","78","75"

from glob import glob
import pandas as pd
import dask.bag as db

files = glob('/path/to/your/csvs/*.csv')
bag = db.from_sequence(files).map(pd.read_csv)

到目前为止我所做的尝试

import pandas as pd
import dask.bag as db
import dask.dataframe as dd

# Create a Dask bag of pandas dataframes
bag = db.from_sequence(list_of_files).map(my_reader_function)

df = bag.map(lambda x: x.to_records()).to_dataframe() # this doesn't work
df = bag.map(lambda x: x.to_dict(orient = <any option>)).to_dataframe() # neither does this

# This gets me really close. It's a bag of Dask DataFrames. 
# But I can't figure out how to concatenate them together
df = bag.map(dd.from_pandas, npartitions = 1)

df = dd.from_delayed(bag) # returns an error

【问题讨论】：

您提供的示例数据可以轻松转换为 pandas 数据框，保存为不同的文件，后者可以读入单个 dask 数据框。因此，不清楚为什么需要使用 dask bag 并将其转换为 dask 数据框。
你问了一个好问题。我应该澄清一下：我需要读取数百万个这些文件，每个文件从几 KB 到几 MB 不等。因此，我没有空间来创建文件的 dask 可读副本。
如果您的自定义读取函数输出单个 pandas 数据帧，您不能将输出转换为 dask 数据帧，并继续这样做并附加到 dask 数据帧吗？
这确实非常接近我想要的！但是，我还没有找到一种不会同时将所有文件读入内存的方法。 bag = db.from_sequence(files).map(pd.read_csv).map(dd.from_pandas, npartitions=1) 然后dd.concat(bag.compute()) 确实可以工作；但它将所有内容读入内存，这对数百万个文件不起作用

标签： python pandas dataframe dask

【解决方案1】：

我建议将 dask.delayed 与 dask.dataframe 一起使用。有一个很好的例子可以做你想做的事情：

https://docs.dask.org/en/latest/delayed-collections.html

【讨论】：

这正是我想要的！不敢相信我以前没有看过这个页面。谢谢！

【解决方案2】：

这里有另外两个可能的解决方案：

1.将包转换为数据框列表，然后使用dd.multi.concat:

bag #a dask bag of dataframes
list_of_dfs = bag.compute()
df = dd.multi.concat(list_of_dfs).compute()

2。转换成一包字典，使用bag.to_dataframe：

bag_of_dicts = bag.map(lambda df: df.to_dict(orient='records')).flatten()
df = bag_of_dicts.to_dataframe().compute()

在我自己的特定用例中，选项 #2 的性能优于选项 #1。

【讨论】：

【解决方案3】：

如果您已经有一袋数据框，那么您可以执行以下操作：

将包转换为延迟分区，
通过连接将延迟的分区转换为数据帧的延迟，
根据这些延迟创建数据帧。

在python代码中：

def bag_to_dataframe(bag, **concat_kwargs):
    partitions = bag.to_delayed()
    dataframes = map(
        dask.delayed(lambda partition: pandas.concat(partition, **concat_kwargs)),
        partitions
    )
    return dask.dataframe.from_delayed(dataframes)

您可能想要控制分区的连接，例如忽略索引。

【讨论】：