Python每日更新DataFrame：如何高效对比当日与昨日数据变更

一、更高效的DataFrame对比实现 #

你手写的对比逻辑可以大幅简化，pandas内置的df.equals()方法已能完成全量一致性校验，它会自动检查行、列、值（包括NaN）是否完全匹配，比逐列判断更简洁高效。同时，差异数据的计算也可以优化得更可靠：

import pandas as pd
import numpy as np
pd.options.mode.chained_assignment = None
import os

# 从数据库读取当日数据（此处为示例）
today_df = pd.DataFrame(columns=["id", "date", "count"])

# 读取昨日数据（存储方式见下文）
yesterday_df = pd.DataFrame(columns=["id", "date", "count"])

# 简化对比逻辑：重置索引避免索引差异导致误判
def is_dataFrames_equal(today_df, yesterday_df):
    return today_df.reset_index(drop=True).equals(yesterday_df.reset_index(drop=True))

if is_dataFrames_equal(today_df, yesterday_df):
    print('no new data found.')
    exit()

# 更可靠的差异计算：用merge定位新增/变更行
diff_df = pd.merge(today_df, yesterday_df, on=["id", "date"], how="left", suffixes=("", "_y"))
# 筛选出count变更或新增的行
diff_df = diff_df[(diff_df["count"] != diff_df["count_y"]) | diff_df["count_y"].isna()]
# 清理临时列
diff_df = diff_df.drop(columns=["count_y"])

df = today_df.copy(deep=True)

说明：

• reset_index(drop=True)是为了避免因每日数据索引不连续，导致equals()误判为不一致
• 用merge计算差异比isin更精准，能覆盖部分字段变更、新增行等场景，不会漏判

二、保存昨日DataFrame的最佳方式 #

根据数据规模和使用场景，推荐以下几种存储方案：

1. Pickle（小数据首选） #

Pickle是Python二进制序列化格式，能完整保留DataFrame的结构、数据类型（如datetime、category），读写速度快，适合中小规模数据：

# 当日数据处理完成后，保存为昨日数据
today_df.to_pickle("yesterday_data.pkl")

# 次日读取昨日数据
if os.path.exists("yesterday_data.pkl"):
    yesterday_df = pd.read_pickle("yesterday_data.pkl")
else:
    # 首次运行无历史数据，直接视为有变更
    yesterday_df = pd.DataFrame(columns=["id", "date", "count"])

2. Feather/Parquet（大数据首选） #

这两种是面向列的高效存储格式，压缩比高、读写速度快，适合百万行以上的大数据，同时保留完整数据类型：

# Feather示例
today_df.to_feather("yesterday_data.feather")
yesterday_df = pd.read_feather("yesterday_data.feather")

# Parquet示例（需先安装pyarrow或fastparquet）
today_df.to_parquet("yesterday_data.parquet")
yesterday_df = pd.read_parquet("yesterday_data.parquet")

3. CSV（兼容性优先） #

如果需要和其他工具共享数据，CSV是通用格式，但会丢失数据类型（如日期转为字符串），需读取时手动转换：

today_df.to_csv("yesterday_data.csv", index=False)
yesterday_df = pd.read_csv("yesterday_data.csv", parse_dates=["date"])

4. 数据库快照（长期存储/多机器共享） #

如果数据本身存在数据库中，可以新增一张data_snapshot表，用日期标识存储每日快照：

from sqlalchemy import create_engine
engine = create_engine("你的数据库连接字符串")

# 读取昨日快照
yesterday_date = pd.Timestamp.today() - pd.Timedelta(days=1)
yesterday_df = pd.read_sql(f"SELECT * FROM data_snapshot WHERE snapshot_date = '{yesterday_date.date()}'", engine)

# 处理完成后写入当日快照
today_df["snapshot_date"] = pd.Timestamp.today().date()
today_df.to_sql("data_snapshot", engine, if_exists="append", index=False)

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