连接dask数据帧和pandas数据帧

Question

我有一个dask数据帧（df），大约有2.5亿行（来自一个10Gb的CSV文件）。我有另一个包含25000行的pandas数据帧（ndf）。我想将pandas dataframe的第一列添加到dask dataframe中，每个项目重复10000次。在

这是我试过的密码。我已经把问题缩小了。在

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

pd.DataFrame(np.random.rand(25000, 2)).to_csv("tempfile.csv")
df = dd.read_csv("tempfile.csv")
ndf = pd.DataFrame(np.random.randint(1000, 3500, size=2500))
df['Node'] = np.repeat(ndf[0], 10)

有了这段代码，我最终得到了一个错误。在

ValueError: Not all divisions are known, can't align partitions. Please use set_index to set the index.

我可以执行reset_index()，然后执行set_index()，为dask数据帧生成df.known_divisionsTrue。但这是一项耗时的手术。有没有更好的更快的方法来做我想做的事？我能用熊猫来做这个吗？在

最终目标是从ndf中查找行，其中df中的任何对应行都符合某些条件。在

Answer 1

使用相同的工作流，您可以按照建议手动设置divisions}，here

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

pd.DataFrame(np.random.rand(25000, 2)).to_csv("tempfile.csv", index=False)
df = dd.read_csv("tempfile.csv")
ndf = pd.DataFrame(np.random.randint(1000, 3500, size=2500))


df.divisions = (0, len(df)-1)
df["Note"] = dd.from_array(np.repeat(ndf.values, 10))

我不认为使用np.repeat是非常有效的，尤其是对于大df。在

Answer 2

您的基本算法是“我希望将df['Node']的前10个值设置为ndf的第一个值，将后10个值设置为ndf的下一个值，依此类推”。这在Dask中很难实现，是因为它不知道每个分区中有多少行：您正在从csv读取数据，以X字节表示的行数取决于每个部分中的数据。其他格式可以提供更多信息。。。在

因此，您肯定需要对数据进行两次传递。您可以使用索引来计算除法并进行排序。在我看来，你能做的最简单的事情就是测量除法的长度，然后得到每个除法起点的偏移量：

lengths = df.map_partitions(len).compute()
offsets = np.cumsum(lengths.values)
offsets -= offsets[0]

现在使用自定义延迟函数来处理部件

^{pr2}$