将大型 Dask 数据框与小型 Pandas 数据框合并

Posted 2023-02-16

【中文标题】将大型 Dask 数据框与小型 Pandas 数据框合并

【英文标题】：Merge a large Dask dataframe with a small Pandas dataframe

【发布时间】：2017-01-21 01:45:01

【问题描述】：

按照此处的示例：YouTube: Dask-Pandas Dataframe Join 我尝试将 ~70GB Dask 数据帧与我作为 Pandas 数据帧加载的 ~24MB 合并。

合并在 A 和 B 两列上，我没有设置任何索引：

import dask.dataframe as dd
from dask.diagnostics import ProgressBar

small_df = pd.read_csv(dataframe1) # as pandas
large_df = dd.read_csv(dataframe2) #as dask.dataframe

df2 = large_df.merge(small_df, how='left', left_on=leftcolumns, right_on=rightcolumns) #do the merge

A = df2[df2['some column'] == 'somevalue'] #do a reduction that would fit on my computer

pbar = ProgressBar()
pbar.register()

result = A.compute()

我使用的是具有 16GB RAM 和 4 核的 Windows 计算机。 我使用进度条来评估合并过程的进度。我昨晚整晚都离开了它。我今天早上重新启动它，到目前为止大约半小时，进度为 0%。

谢谢你，感谢你的帮助，

更新

我在配备 8GB RAM 的 Mac 上进行了尝试，效果非常好。我相信我有 Anaconda 附带的 Dask 发行版。无论如何，我认为我没有做任何不同的事情。

我按照上述编码（21 分钟）分享我的结果和时间：

In [26]: C = result1.compute()
[########################################] | 100% Completed | 21min 13.4s
[########################################] | 100% Completed | 21min 13.5s
[########################################] | 100% Completed | 21min 13.6s
[########################################] | 100% Completed | 21min 13.6s

更新 2

我在我的 Windows 计算机上更新到最新版本的 Dask，它运行良好。

【问题讨论】：

只是一个旁注，在布尔索引中使用df 而不是df2 是错误的吗？ A = df2[df['some column'] == 'somevalue']

哦，抱歉，我才看到你的意思。是的，这是一个错误。我修好了它。谢谢！

@dleal，既然这个问题已经解决，您能否提供解决方案作为答案并接受它（即使这是您自己的问题，也鼓励），将其从未回答的列表中删除？谢谢！

要添加到@Garrett 的评论：请具体说明您拥有的 Dask 版本以及您刚刚更新到什么。

您也可以使用 joblib，并在多个内核上运行该进程。一般的想法是将大型数据集拆分为一组较小的数据集（即[large_df_part1, ... , large_df_partN]，然后使用 joblib 将它们分配给不同的处理器。

【参考方案1】：

您可以迭代唯一的相等值并使用循环分配其他列：

unioun_set = list(set(small_df['common_column']) & set(large_df['common_column']))
for el in union_set:
    for column in small_df.columns:
        if column not in large_df.columns:
            large_df.loc[large_df['common_column'] == el,column] = small_df.loc[small_df['common_column'] ==  el,column]

【参考方案2】：

在处理大数据时，对数据进行分区非常重要，同时必须拥有足够的集群和内存大小。

您可以尝试使用spark。

DASK 是一个纯 Python 框架，它做更多相同的事情，即它允许人们在本地或集群上运行相同的 Pandas 或 NumPy 代码。然而，Apache Spark 带来了一个涉及新 API 和执行模型的学习曲线，尽管使用了 Python 包装器。

您可以尝试对数据进行分区并将其存储到parquet 文件中。