使用Pandas提取CSV新增数据遇异常，求技术解决方案

需要让new_data仅包含脚本本次运行时获取的新增数据——比如unique_seller_data.csv现有数据是[1,2,3,4]，历史数据previous_data是[1,2,3]，那么new_data应该只保留[4]。

此前尝试的多段代码均未实现预期效果，使用merge方法时还触发了类型不匹配错误：

line 1387, in _maybe_coerce_merge_keys
raise ValueError(msg)
ValueError: You are trying to merge on object and float64 columns. If you wish to proceed you should use pd.concat

已尝试的代码片段 #

第一段代码 #

deduplicated_data.to_csv('unique_seller_data.csv', index=False)

    try:
        previous_data = pd.read_csv(f'unique_seller_data{last_date_str}.csv')
        new_data = sellers_data[~sellers_data.isin(previous_data)].dropna()
        new_data.to_csv(f'new_data_added_{date_str}.csv', index=False)
        
    except FileNotFoundError:
        print('no old data to compare to')
        deduplicated_data.to_csv(f'new_data_added_{date_str}.csv', index=False)
        print(f'{len(deduplicated_data)} rows added to new_data_added.csv')

第二段代码 #

deduplicated_data.to_csv('unique_seller_data.csv', index=False)

    try:
        previous_data = pd.read_csv(f'new_data_added_{last_date_str}.csv')
        new_data = deduplicated_data[~deduplicated_data.isin(previous_data)].dropna()
        new_data.to_csv(f'new_data_added_{date_str}.csv', index=False)
    except FileNotFoundError:
        print('no old data to compare to')
        deduplicated_data.to_csv(f'new_data_added_{date_str}.csv', index=False)
    print('formatted')

解决方案 #

问题根源 #

1. 原代码用isin整行匹配时，极易因CSV读取时的自动类型转换差异（比如数字转字符串、空值转float）导致匹配失效，无法正确过滤已有数据。
2. merge报错是因为关联键的列类型不统一（object和float64），需先统一类型再做对比。

修正方案 #

方案1：基于唯一标识列过滤（推荐）

如果数据中有唯一标识字段（比如seller_id、店铺ID），用它判断新增数据最可靠：

deduplicated_data.to_csv('unique_seller_data.csv', index=False)

try:
    # 读取历史数据，统一唯一标识列类型
    previous_data = pd.read_csv(f'unique_seller_data{last_date_str}.csv')
    # 将唯一标识列强制转为相同类型（这里以字符串为例）
    deduplicated_data['seller_id'] = deduplicated_data['seller_id'].astype(str)
    previous_data['seller_id'] = previous_data['seller_id'].astype(str)
    
    # 过滤出历史数据中没有的新增行
    new_data = deduplicated_data[~deduplicated_data['seller_id'].isin(previous_data['seller_id'])]
    
    new_data.to_csv(f'new_data_added_{date_str}.csv', index=False)
    print(f'{len(new_data)} rows added to new_data_added.csv')

except FileNotFoundError:
    print('no old data to compare to')
    deduplicated_data.to_csv(f'new_data_added_{date_str}.csv', index=False)
    print(f'{len(deduplicated_data)} rows added to new_data_added.csv')

方案2：无唯一标识时用整行哈希匹配

如果没有单独的唯一标识列，可给每行生成哈希值来判断是否重复：

import hashlib

def generate_row_hash(row):
    # 将行数据转为固定格式的字符串，生成哈希值
    row_str = '|'.join([str(val) for val in row.values])
    return hashlib.md5(row_str.encode()).hexdigest()

deduplicated_data.to_csv('unique_seller_data.csv', index=False)

try:
    previous_data = pd.read_csv(f'unique_seller_data{last_date_str}.csv')
    
    # 给两行数据生成哈希列
    deduplicated_data['row_hash'] = deduplicated_data.apply(generate_row_hash, axis=1)
    previous_data['row_hash'] = previous_data.apply(generate_row_hash, axis=1)
    
    # 过滤新增行并删除临时哈希列
    new_data = deduplicated_data[~deduplicated_data['row_hash'].isin(previous_data['row_hash'])]
    new_data = new_data.drop(columns=['row_hash'])
    
    new_data.to_csv(f'new_data_added_{date_str}.csv', index=False)
    print(f'{len(new_data)} rows added to new_data_added.csv')

except FileNotFoundError:
    print('no old data to compare to')
    deduplicated_data.to_csv(f'new_data_added_{date_str}.csv', index=False)
    print(f'{len(deduplicated_data)} rows added to new_data_added.csv')

关键说明 #

• 优先使用唯一标识列对比，效率更高且不受列类型小差异影响。
• 整行哈希匹配能规避isin的类型敏感问题，但性能略低于唯一标识列对比。
• 读取CSV后统一关联列类型，是解决merge类型不匹配错误的核心操作。

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