使用 pandas 增量数据加载

Question

我正在尝试使用 pandas 实现增量数据导入。

我有两个数据帧：df_old（原始数据，之前加载）和df_new（新数据，与df_old合并）。

df_old/df_new 中的数据在多个列上是唯一的（为简单起见，我们只说 2：key1 和 key2）。其他列是要合并的数据，可以说，它们也只是其中的 2 列：val1 和 val2。

除此之外，还有一列需要注意：change_id - 它会随着每个新条目覆盖旧条目而增加

导入的逻辑非常简单：

1. 如果 df_new 中有新的密钥对，则应将其（以及相应的 val1/val2 值）添加到 df_old

2. 如果 df_new 中存在 df_old 中存在的密钥对，则：

   2a) 如果 df_old 和 df_new 中的对应值相同，则应保留旧值

   2b) 如果 df_old 和 df_new 中的对应值不同，则 df_new 中的值应替换 df_old 中的旧值

3. 无需关心 dala 删除（如果 df_old 中存在某些数据，但 df_new 中不存在）

到目前为止，我想出了两种不同的解决方案：

>>> df_old = pd.DataFrame([['A1','B2',1,2,1],['A1','A2',1,3,1],['B1','A2',1,3,1],['B1','B2',1,4,1],], columns=['key1','key2','val1','val2','change_id'])
>>> df_old.set_index(['key1','key2'], inplace=True)
>>> df_old

           val1  val2  change_id
key1 key2                       
A1   B2       1     2          1
     A2       1     3          1
B1   A2       1     3          1
     B2       1     4          1

>>> df_new = pd.DataFrame([['A1','B2',2,1,2],['A1','A2',1,3,2],['C1','B2',2,1,2]], columns=['key1','key2','val1','val2','change_id'])
>>> df_new.set_index(['key1','key2'], inplace=True)
>>> df_new

           val1  val2  change_id
key1 key2                       
A1   B2       2     1          2
     A2       1     3          2
C1   B2       2     1          2

解决方案1

# this solution groups concatenated old data with new ones, group them by keys and for each group evaluates if new data are different
def merge_new(x):    
    if x.shape[0] == 1:
        return x.iloc[0]
    else: 
        if x.iloc[0].loc[['val1','val2']].equals(x.iloc[1].loc[['val1','val2']]):
            return x.iloc[0]
        else:
            return x.iloc[1]

def solution1(df_old, df_new):
    merged = pd.concat([df_old, df_new]) 
    return merged.groupby(level=['key1','key2']).apply(merge_new).reset_index()

解决方案2

# this solution uses pd.merge to merge data + additional logic to compare merged rows and select new data
>>> def solution2(df_old, df_new):
>>>    merged = pd.merge(df_old, df_new, left_index=True, right_index=True, how='outer', suffixes=('_old','_new'), indicator='ind')
>>>    merged['isold'] = (merged.loc[merged['ind'] == 'both',['val1_old','val2_old']].rename(columns=lambda x: x[:-4]) == merged.loc[merged['ind'] == 'both',['val1_new','val2_new']].rename(columns=lambda x: x[:-4])).all(axis=1)
>>>    merged.loc[merged['ind'] == 'right_only','isold'] = False    
>>>    merged['isold'] = merged['isold'].fillna(True)
>>>    return pd.concat([merged[merged['isold'] == True][['val1_old','val2_old','change_id_old']].rename(columns=lambda x: x[:-4]), merged[merged['isold'] == False][['val1_new','val2_new','change_id_new']].rename(columns=lambda x: x[:-4])])

>>> solution1(df_old, df_new)

  key1 key2  val1  val2  change_id
0   A1   A2     1     3          1
1   A1   B2     2     1          2
2   B1   A2     1     3          1
3   B1   B2     1     4          1
4   C1   B2     2     1          2


>>> solution2(df_old, df_new)

           val1  val2  change_id
key1 key2                       
A1   A2     1.0   3.0        1.0
B1   A2     1.0   3.0        1.0
     B2     1.0   4.0        1.0
A1   B2     2.0   1.0        2.0
C1   B2     2.0   1.0        2.0

然而，这两项工作我仍然对巨大数据帧上的性能感到非常失望。问题是：有没有更好的方法来做到这一点？任何有关速度提高的提示都将非常受欢迎......

>>> %timeit solution1(df_old, df_new)
100 loops, best of 3: 10.6 ms per loop

>>> %timeit solution2(df_old, df_new)
100 loops, best of 3: 14.7 ms per loop

Answer 1

这是一种非常快的方法：

merged = pd.concat([df_old.reset_index(), df_new.reset_index()])
merged = merged.drop_duplicates(["key1", "key2", "val1", "val2"]).drop_duplicates(["key1", "key2"], keep="last")
# 100 loops, best of 3: 1.69 ms per loop

#   key1 key2  val1  val2  change_id
# 1   A1   A2     1     3          1
# 2   B1   A2     1     3          1
# 3   B1   B2     1     4          1
# 0   A1   B2     2     1          2
# 2   C1   B2     2     1          2

这里的基本原理是连接所有行并简单地调用drop_duplicates两次，而不是依赖连接逻辑来获取所需的行。第一次调用会drop_duplicates删除键和值列上源自df_new该匹配的行，因为此方法的默认行为是保留第一个重复行（在本例中为来自 的行df_old）。第二个调用删除与键列匹配的重复项，但指定last应保留每组重复项的行。

此方法假设行按 ; 排序change_id。考虑到示例 DataFrame 的连接顺序，这是一个安全的假设。但是，如果这是对您的真实数据的错误假设，只需在删除重复项之前.sort_values('change_id')调用即可。merged