本文详解如何在 pandas 中安全、可靠地实现基于唯一键（如 email）的增量更新，避免重复写入，涵盖去重逻辑优化、索引对齐、原子写入及线程安全建议。
本文详解如何在 pandas 中安全、可靠地实现基于唯一键（如 email）的增量更新，避免重复写入，涵盖去重逻辑优化、索引对齐、原子写入及线程安全建议。
在日志统计、用户行为追踪等场景中，常需以邮箱（email）为唯一标识持续更新成功/失败计数。但直接读取 → 判断 → 修改 → 写入的流程，若缺乏原子性或逻辑疏漏，极易导致重复行（如多次写入同一 email），破坏数据一致性。原代码虽尝试用 email in df["email"].values 检查存在性，但仍出错——根本原因在于：该判断未处理空 DataFrame 边界情况，且 .append()（已弃用）与重复索引查找逻辑存在竞态风险；更重要的是，它未对写入前的最终 DataFrame 做强制去重保障。
✅ 正确解法不是“事后补救”，而是构建健壮的主键驱动更新机制。推荐以下三步法：
1. 使用 set_index + combine_first 实现幂等更新（推荐）
这是最符合 Pandas 设计哲学的方式：将 email 设为索引，构造新数据帧，利用 combine_first 自动按索引合并并保留非空值。
import pandas as pd
from datetime import datetime
def save_data(email: str, is_success: int, is_failed: int) -> None:
csv_file_path = "data.csv"
# 读取现有数据（若不存在则创建空结构）
try:
df = pd.read_csv(csv_file_path).set_index("email")
except FileNotFoundError:
df = pd.DataFrame(columns=["success_count", "failure_count", "last_updated_on"]).set_index("email")
# 构造单行新数据（确保索引为 email）
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
new_row = pd.DataFrame({
"success_count": [is_success],
"failure_count": [is_failed],
"last_updated_on": [now]
}, index=pd.Index([email], name="email"))
# 关键：按索引合并，新值覆盖旧值（若字段非空），自动处理新增/更新
df = df.combine_first(new_row)
# 若 email 已存在，其 success/failure 字段会被累加？否！combine_first 不支持累加，需手动处理：
if email in df.index:
df.loc[email, "success_count"] += is_success
df.loc[email, "failure_count"] += is_failed
df.loc[email, "last_updated_on"] = now
else:
df = pd.concat([df, new_row]) # 安全追加
# 重置索引并写入（含去重兜底）
df = df.reset_index()
df.to_csv(csv_file_path, index=False)2. 更简洁可靠的方案：pd.concat + drop_duplicates
若偏好清晰控制流，可统一用 concat 合并，再按 email 去重并聚合：
def save_data_v2(email: str, is_success: int, is_failed: int) -> None:
csv_file_path = "data.csv"
# 读取或初始化
try:
df = pd.read_csv(csv_file_path)
except FileNotFoundError:
df = pd.DataFrame(columns=["email", "success_count", "failure_count", "last_updated_on"])
# 构建新记录
new_df = pd.DataFrame([{
"email": email,
"success_count": is_success,
"failure_count": is_failed,
"last_updated_on": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}])
# 合并并去重：保留最新时间戳的记录，同时累加计数
combined = pd.concat([df, new_df], ignore_index=True)
# 按 email 分组，对计数列求和，对 last_updated_on 取最大值（即最新时间）
aggregated = combined.groupby("email", as_index=False).agg({
"success_count": "sum",
"failure_count": "sum",
"last_updated_on": "max" # 确保时间戳为最后一次更新
})
aggregated.to_csv(csv_file_path, index=False)⚠️ 关键注意事项
- 弃用 .append()：Pandas 2.0+ 已移除，必须改用 pd.concat()。
- 避免 in df["col"].values 性能陷阱：大数据集下效率低，应优先用 email in df.index（索引查询 O(1)）或 df["email"].isin([email]).any()。
- 并发写入风险：上述方法不适用于多进程/多线程高频写入。生产环境建议改用 SQLite 或加文件锁（如 portalocker）。
- 兜底去重：即使逻辑正确，也建议在 to_csv 前加 df.drop_duplicates(subset=["email"], keep="last") 作为安全屏障。
- 时间格式统一：使用 datetime.now().isoformat() 或固定格式字符串，避免解析歧义。
✅ 最终验证
运行原测试数据后，data.csv 将严格包含 3 行唯一 email，且 success_count 与 failure_count 分别为 (1,1), (1,1), (1,1)，last_updated_on 为各自最后更新时间——完全满足业务预期。
掌握基于主键的聚合更新范式，远比简单调用 drop_duplicates() 更具扩展性与鲁棒性。
