如何在增量管道中保存历史输入并检测数据行变更？

要实现这个需求，核心是持久化存储上一次的DataFrame，每次新数据输入时先读取历史版本，再将当前数据保存为下一次的历史。下面分不同场景给出具体实现：

1. 本地文件存储（适合小规模离线数据） #

用高效的文件格式（比如Parquet）保存历史DataFrame，每次处理新数据时先加载历史文件，处理完后覆盖保存当前数据。

示例代码（基于Pandas）：

import pandas as pd
import os

HISTORY_PATH = "last_dataframe.parquet"

def process_new_data(current_df):
    # 加载上一次的DataFrame（如果存在）
    if os.path.exists(HISTORY_PATH):
        last_df = pd.read_parquet(HISTORY_PATH)
    else:
        last_df = pd.DataFrame(columns=current_df.columns)
    
    # 对比行变更：找出新增和删除的行
    # 假设用主键列（比如'id'）来匹配行
    merged = pd.merge(last_df, current_df, on='id', how='outer', indicator=True)
    added_rows = merged[merged['_merge'] == 'right_only'].drop('_merge', axis=1)
    removed_rows = merged[merged['_merge'] == 'left_only'].drop('_merge', axis=1)
    
    # 保存当前DataFrame作为下一次的历史
    current_df.to_parquet(HISTORY_PATH, index=False)
    
    return last_df, added_rows, removed_rows

# 测试使用
current_data = pd.DataFrame({'id': [1,2,3], 'value': ['a','b','c']})
last_df, added, removed = process_new_data(current_data)

2. 内存缓存（适合实时会话/短生命周期管道） #

如果你的管道是在单个Python会话内运行（比如实时数据流处理），可以用类属性缓存历史DataFrame，避免频繁IO操作。

示例代码：

import pandas as pd

class DeltaPipeline:
    def __init__(self):
        self.last_df = pd.DataFrame()
    
    def process(self, current_df):
        # 对比行变更
        merged = pd.merge(self.last_df, current_df, on='id', how='outer', indicator=True)
        added = merged[merged['_merge'] == 'right_only'].drop('_merge', axis=1)
        removed = merged[merged['_merge'] == 'left_only'].drop('_merge', axis=1)
        
        # 更新历史DataFrame
        self.last_df = current_df.copy()
        
        return self.last_df, added, removed

# 测试使用
pipeline = DeltaPipeline()
current_data1 = pd.DataFrame({'id': [1,2], 'value': ['a','b']})
last, added, removed = pipeline.process(current_data1)

current_data2 = pd.DataFrame({'id': [2,3], 'value': ['b','c']})
last, added, removed = pipeline.process(current_data2)
# 此时added是id=3的行，removed是id=1的行

3. 数据库存储（适合大规模/分布式场景） #

如果数据量较大或需要分布式访问，将历史DataFrame存在数据库表中，每次处理新数据时先读取表中数据，再更新表。

示例代码（基于SQLite）：

import pandas as pd
import sqlite3

DB_PATH = "delta_pipeline.db"
TABLE_NAME = "last_dataframe"

def get_last_df():
    conn = sqlite3.connect(DB_PATH)
    try:
        last_df = pd.read_sql(f"SELECT * FROM {TABLE_NAME}", conn)
    except pd.io.sql.DatabaseError:
        # 表不存在时返回空DataFrame
        last_df = pd.DataFrame()
    conn.close()
    return last_df

def save_current_df(current_df):
    conn = sqlite3.connect(DB_PATH)
    # 覆盖写入表
    current_df.to_sql(TABLE_NAME, conn, if_exists='replace', index=False)
    conn.close()

def process_new_data(current_df):
    last_df = get_last_df()
    
    # 对比行变更
    merged = pd.merge(last_df, current_df, on='id', how='outer', indicator=True)
    added = merged[merged['_merge'] == 'right_only'].drop('_merge', axis=1)
    removed = merged[merged['_merge'] == 'left_only'].drop('_merge', axis=1)
    
    save_current_df(current_df)
    
    return last_df, added, removed

关键注意点 #

• 主键选择：必须有唯一标识行的主键（比如id），否则无法准确对比行的新增/删除；如果没有主键，可以考虑用所有列的组合作为匹配依据，但效率较低。
• 数据格式兼容性：保存历史DataFrame时，要确保格式能保留所有数据类型（Parquet比CSV更适合，因为CSV会丢失类型信息）。
• 并发安全：如果是多进程/多线程场景，本地文件或数据库存储要注意加锁，避免读写冲突。

内容的提问来源于stack exchange，提问作者Max Magid