问题概述
简短版
如何将Dask Bag of Pandas DataFrames转换为单个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
bag = db.from_sequence(files).map(pd.read_csv).map(dd.from_pandas, npartitions=1)
然后dd.concat(bag.compute())
确实可以工作;但它会将所有内容都读入内存,这对于数百万个文件是行不通的。 - natemcintosh