合并多个大型 DataFrame 的有效方法

Question

假设我有 4 个小的 DataFrames

df1、df2、df3 和 df4

import pandas as pd
from functools import reduce
import numpy as np

df1 = pd.DataFrame([['a', 1, 10], ['a', 2, 20], ['b', 1, 4], ['c', 1, 2], ['e', 2, 10]])
df2 = pd.DataFrame([['a', 1, 15], ['a', 2, 20], ['c', 1, 2]])
df3 = pd.DataFrame([['d', 1, 10], ['e', 2, 20], ['f', 1, 1]])  
df4 = pd.DataFrame([['d', 1, 10], ['e', 2, 20], ['f', 1, 15]])   


df1.columns = ['name', 'id', 'price']
df2.columns = ['name', 'id', 'price']
df3.columns = ['name', 'id', 'price']    
df4.columns = ['name', 'id', 'price']   

df1 = df1.rename(columns={'price':'pricepart1'})
df2 = df2.rename(columns={'price':'pricepart2'})
df3 = df3.rename(columns={'price':'pricepart3'})
df4 = df4.rename(columns={'price':'pricepart4'})

上面创建的是4个DataFrame，我想要的在下面的代码中。

# Merge dataframes
df = pd.merge(df1, df2, left_on=['name', 'id'], right_on=['name', 'id'], how='outer')
df = pd.merge(df , df3, left_on=['name', 'id'], right_on=['name', 'id'], how='outer')
df = pd.merge(df , df4, left_on=['name', 'id'], right_on=['name', 'id'], how='outer')

# Fill na values with 'missing'
df = df.fillna('missing')

所以我已经为没有很多行和列的 4 个 DataFrame 实现了这一点。

基本上，我想将上述外部合并解决方案扩展到大小为 62245 X 3 的 MULTIPLE (48) DataFrames：

所以我从另一个使用 lambda reduce 的 StackOverflow 答案构建，提出了这个解决方案：

from functools import reduce
import pandas as pd
import numpy as np
dfList = []

#To create the 48 DataFrames of size 62245 X 3
for i in range(0, 49):

    dfList.append(pd.DataFrame(np.random.randint(0,100,size=(62245, 3)), columns=['name',  'id',  'pricepart' + str(i + 1)]))


#The solution I came up with to extend the solution to more than 3 DataFrames
df_merged = reduce(lambda  left, right: pd.merge(left, right, left_on=['name', 'id'], right_on=['name', 'id'], how='outer'), dfList).fillna('missing')

这导致MemoryError。

我不知道该怎么做才能阻止内核死机..我已经坚持了两天..我执行的 EXACT 合并操作的一些代码不会导致 MemoryError 或能给你同样结果的东西，将不胜感激。

此外，主 DataFrame 中的 3 列（不是示例中可重现的 48 个 DataFrame）的类型为 int64、int64 和 float64，我希望它们保持这种状态，因为整数和它所代表的浮动。

编辑：

我没有反复尝试运行合并操作或使用 reduce lambda 函数，而是以 2 个一组来完成它！另外，我更改了一些列的数据类型，有些不需要是float64。所以我把它归结为float16。它走得很远，但最终还是抛出了MemoryError。

intermediatedfList = dfList    

tempdfList = []    

#Until I merge all the 48 frames two at a time, till it becomes size 2
while(len(intermediatedfList) != 2):

    #If there are even number of DataFrames
    if len(intermediatedfList)%2 == 0:

        #Go in steps of two
        for i in range(0, len(intermediatedfList), 2):

            #Merge DataFrame in index i, i + 1
            df1 = pd.merge(intermediatedfList[i], intermediatedfList[i + 1], left_on=['name',  'id'], right_on=['name',  'id'], how='outer')
            print(df1.info(memory_usage='deep'))

            #Append it to this list
            tempdfList.append(df1)

        #After DataFrames in intermediatedfList merging it two at a time using an auxillary list tempdfList, 
        #Set intermediatedfList to be equal to tempdfList, so it can continue the while loop. 
        intermediatedfList = tempdfList 

    else:

        #If there are odd number of DataFrames, keep the first DataFrame out

        tempdfList = [intermediatedfList[0]]

        #Go in steps of two starting from 1 instead of 0
        for i in range(1, len(intermediatedfList), 2):

            #Merge DataFrame in index i, i + 1
            df1 = pd.merge(intermediatedfList[i], intermediatedfList[i + 1], left_on=['name',  'id'], right_on=['name',  'id'], how='outer')
            print(df1.info(memory_usage='deep'))
            tempdfList.append(df1)

        #After DataFrames in intermediatedfList merging it two at a time using an auxillary list tempdfList, 
        #Set intermediatedfList to be equal to tempdfList, so it can continue the while loop. 
        intermediatedfList = tempdfList 

有什么办法可以优化我的代码以避免MemoryError，我什至使用了 AWS 192GB RAM（我现在欠他们 7 美元，我本可以给你一个），这比我更远'已经得到了，在将 28 个 DataFrame 的列表减少到 4 个之后，它仍然抛出 MemoryError ..

Answer 1

您可能会从使用pd.concat 执行索引对齐连接中获得一些好处。这也应该比外部合并更快，内存效率更高。

df_list = [df1, df2, ...]
for df in df_list:
    df.set_index(['name', 'id'], inplace=True)

df = pd.concat(df_list, axis=1) # join='inner'
df.reset_index(inplace=True)

或者，您可以将concat（第二步）替换为迭代的join：

from functools import reduce
df = reduce(lambda x, y: x.join(y), df_list)

这可能会或可能不会比merge更好。

Answer 2

似乎是 dask 数据帧设计的一部分（数据帧内存不足）。看 Best way to join two large datasets in Pandas 例如代码。很抱歉没有复制和粘贴，但我不想让我在链接条目中试图从回答者那里获得荣誉。

Answer 3

您可以尝试一个简单的for 循环。我应用的唯一内存优化是通过pd.to_numeric 向下转换为最优化的int 类型。

我还使用字典来存储数据帧。这是保存可变数量变量的好习惯。

import pandas as pd

dfs = {}
dfs[1] = pd.DataFrame([['a', 1, 10], ['a', 2, 20], ['b', 1, 4], ['c', 1, 2], ['e', 2, 10]])
dfs[2] = pd.DataFrame([['a', 1, 15], ['a', 2, 20], ['c', 1, 2]])
dfs[3] = pd.DataFrame([['d', 1, 10], ['e', 2, 20], ['f', 1, 1]])  
dfs[4] = pd.DataFrame([['d', 1, 10], ['e', 2, 20], ['f', 1, 15]])   

df = dfs[1].copy()

for i in range(2, max(dfs)+1):
    df = pd.merge(df, dfs[i].rename(columns={2: i+1}),
                  left_on=[0, 1], right_on=[0, 1], how='outer').fillna(-1)
    df.iloc[:, 2:] = df.iloc[:, 2:].apply(pd.to_numeric, downcast='integer')

print(df)

   0  1   2   3   4   5
0  a  1  10  15  -1  -1
1  a  2  20  20  -1  -1
2  b  1   4  -1  -1  -1
3  c  1   2   2  -1  -1
4  e  2  10  -1  20  20
5  d  1  -1  -1  10  10
6  f  1  -1  -1   1  15

通常，您不应将诸如“missing”之类的字符串与数字类型组合，因为这会将您的整个系列变成object 类型系列。这里我们使用-1，但您可能希望将NaN 与float dtype 一起使用。

Answer 4

因此，您有 48 个 df，每个 3 列 - 名称、id 和每个 df 的不同列。

你不必使用合并......

相反，如果你连接所有的 dfs

df = pd.concat([df1,df2,df3,df4])

您将收到：

Out[3]: 
   id name  pricepart1  pricepart2  pricepart3  pricepart4
0   1    a        10.0         NaN         NaN         NaN
1   2    a        20.0         NaN         NaN         NaN
2   1    b         4.0         NaN         NaN         NaN
3   1    c         2.0         NaN         NaN         NaN
4   2    e        10.0         NaN         NaN         NaN
0   1    a         NaN        15.0         NaN         NaN
1   2    a         NaN        20.0         NaN         NaN
2   1    c         NaN         2.0         NaN         NaN
0   1    d         NaN         NaN        10.0         NaN
1   2    e         NaN         NaN        20.0         NaN
2   1    f         NaN         NaN         1.0         NaN
0   1    d         NaN         NaN         NaN        10.0
1   2    e         NaN         NaN         NaN        20.0
2   1    f         NaN         NaN         NaN        15.0

现在您可以按名称和 ID 分组并取总和：

df.groupby(['name','id']).sum().fillna('missing').reset_index()

如果您尝试使用 48 dfs，您会发现它解决了 MemoryError：

dfList = []
#To create the 48 DataFrames of size 62245 X 3
for i in range(0, 49):
    dfList.append(pd.DataFrame(np.random.randint(0,100,size=(62245, 3)), columns=['name',  'id',  'pricepart' + str(i + 1)]))

df = pd.concat(dfList)
df.groupby(['name','id']).sum().fillna('missing').reset_index()