如何快速优化同一CSV文件中游戏Loot日志重复行的对比处理

问题背景 #

我有一个包含72万行、10列的CSV文件，与问题相关的列如下：

['timestamp_utc', 'looted_by__name', 'item_id', 'quantity'] 

该文件是游戏中玩家拾取地面物品的日志，存在bug：同一玩家拾取同一物品的记录会被写入两行（两行最多间隔5行），仅timestamp_utc列有细微差异，其余['looted_by__name', 'item_id', 'quantity']列完全相同，示例：

2024-06-23T11:40:43.2187312Z,Georgeeto,T4_SOUL,2
2024-06-23T11:40:43.4588316Z,Georgeeto,T4_SOUL,2

其中，2024-06-23T11:40:43.2187312Z为timestamp_utc，'Georgeeto'为looted_by__name，T4_SOUL为item_id，2为quantity。

需求 #

检测两行的['looted_by__name', 'item_id', 'quantity']是否完全相同，若相同则计算时间戳差值，若差值小于0.5秒：

• 将这两行存入Corrupted.csv
• 仅保留其中一行到Clean.csv

初始实现（耗时6-13小时） #

import pandas as pd
import time
from datetime import datetime

start_time = time.time()
combined_df_3 = pd.read_csv("Processing/combined_file_refined.csv", delimiter= ',', usecols=['timestamp_utc', 'looted_by__name', 'item_id', 'quantity'])
combined_df_4 = pd.read_csv("Processing/combined_file_refined.csv", delimiter= ',', usecols=['timestamp_utc', 'looted_by__name', 'item_id', 'quantity'])
bugged_item_df = pd.DataFrame()
clean_item_df = pd.DataFrame()

bugged_item_list = []
clean_item_list = []

date_format = '%Y-%m-%dT%H:%M:%S.%f'

for index1,row1 in combined_df_3.iterrows():
    n = 0 
    time_stamp_1 = datetime.strptime(row1['timestamp_utc'][:26], date_format)
    name_1 = row1['looted_by__name']
    item_id_1 = row1['item_id']
    quantity_1 = row1['quantity']
    

    for index2, row2 in combined_df_4.iterrows():
        print(str(n))
        n += 1
        if n > 5:
            break

        time_stamp_2 = datetime.strptime(row2['timestamp_utc'][:26], date_format)
        name_2 = row2['looted_by__name']
        item_id_2 = row2['item_id']
        quantity_2 = row2['quantity']



        if time_stamp_1 == time_stamp_2 and name_1 == name_2 and item_id_1 == item_id_2 and quantity_2 == quantity_2:
            break # get out of for loop here

        elif name_1 == name_2 and item_id_1 == item_id_2 and quantity_1 == quantity_2:
            if time_stamp_1 > time_stamp_2:
                date_diff = abs(time_stamp_1 - time_stamp_2)
                date_diff_sec = date_diff.total_seconds()
            
            elif time_stamp_1 < time_stamp_2:
                date_diff = abs(time_stamp_2 - time_stamp_1)
                date_diff_sec = date_diff.total_seconds()

            if date_diff_sec < 0.5:
                bugged_item_df = bugged_item_df._append(row1 ,ignore_index=True)
                bugged_item_df = bugged_item_df._append(row2 ,ignore_index=True) #add both lines into  a csv file and not write 1 of them into the final csv file
            
            elif date_diff_sec > 0.5:
                pass # type line into a csv file normally
        else:
            pass # type line into a csv file normally

bugged_item_df.to_csv("test.csv", index=False)
clean_item_df.to_csv('test2.csv', index=False)

end_time = time.time()
execution_time = end_time - start_time

print(f"Execution time: {execution_time} seconds")

注：代码核心逻辑明确，但未完全完成，处理完整文件需6-13小时，需要优化提升速度。

第一次优化（耗时32分钟） #

感谢AKZ的建议，修正循环错误后，耗时从13.4小时缩短至32分钟：

import time
import pandas as pd
from datetime import datetime

#orgnizing the rows
df = pd.read_csv("processing/combined_file_refined.csv", delimiter= ',', usecols=['timestamp_utc', 'looted_by__name', 'item_id', 'quantity'])
df = df.groupby(['looted_by__name', 'timestamp_utc']).sum().reset_index()
df.to_csv("test.csv", index=False)


bugged_item_df = pd.DataFrame()
clean_item_df = pd.DataFrame()
df1 =pd.read_csv("test.csv", delimiter= ',', usecols=['timestamp_utc', 'looted_by__name', 'item_id', 'quantity'])
date_format = '%Y-%m-%dT%H:%M:%S.%f'
n = 0
num_of_runs = 0

start_time = time.time()

for index1,row1 in df.iterrows():
    num_of_runs += 1
    n += 1
    try:
        row2 = df1.iloc[n]
    except IndexError:
        clean_item_df = clean_item_df._append(row1 ,ignore_index=True)
        break
    time_stamp_1 = datetime.strptime(row1['timestamp_utc'][:26], date_format)
    name_1 = row1['looted_by__name']
    item_id_1 = row1['item_id']
    quantity_1 = row1['quantity']
    
    time_stamp_2 = datetime.strptime(row2['timestamp_utc'][:26], date_format)
    name_2 = row2['looted_by__name']
    item_id_2 = row2['item_id']
    quantity_2 = row2['quantity']

    if name_1 != name_2 or item_id_1 != item_id_2 or quantity_1 != quantity_2:
        #add row 1 to df
        continue
    elif time_stamp_1 > time_stamp_2:
        date_diff_1 = abs(time_stamp_1 - time_stamp_2)
        date_diff_sec_1 = date_diff_1.total_seconds()
        if date_diff_sec_1 < 0.5:
            #donot add row 1 to df and add row 1 and row 2 to bugged item list
            bugged_item_df = bugged_item_df._append(row1 ,ignore_index=True)
            bugged_item_df = bugged_item_df._append(row2 ,ignore_index=True)
            pass
        elif date_diff_sec_1 > 0.5:
            clean_item_df = clean_item_df._append(row1 ,ignore_index=True)
            #add row 1 to df
            continue

    elif time_stamp_1 < time_stamp_2:
        date_diff_2 = abs(time_stamp_2 - time_stamp_1)
        date_diff_sec_2 = date_diff_2.total_seconds()
        if date_diff_sec_2 < 0.5:
            bugged_item_df = bugged_item_df._append(row1 ,ignore_index=True)
            bugged_item_df = bugged_item_df._append(row2 ,ignore_index=True)
            #donot add row 1 to df and add row 1 and row 2 to bugged item list
            pass
        elif date_diff_sec_2 > 0.5:
            clean_item_df = clean_item_df._append(row1 ,ignore_index=True)
            #add row 1 to df
            continue
    

bugged_item_df.to_csv("bugged.csv", index=False)
clean_item_df.to_csv("clean.csv", index=False)

end_time = time.time()
execution_time = end_time - start_time
print(f"Execution time: {execution_time} seconds")

第二次优化（耗时60秒） #

通过直接移除重复行进一步优化，耗时缩短至60秒：

import time
import pandas as pd
from datetime import datetime

#orgnizing the rows
combined_df_3 = pd.read_csv("processing/combined_file_refined.csv", delimiter= ',', usecols=['timestamp_utc', 'looted_by__name', 'item_id', 'quantity'])
combined_df_3 = combined_df_3.groupby(['looted_by__name', 'timestamp_utc']).sum().reset_index()
combined_df_3.to_csv("processing/combined_file_orgnized.csv", index=False)



bugged_item_df = pd.DataFrame()
bugged_item_2df = pd.DataFrame()
combined_df_4 =pd.read_csv("processing/combined_file_orgnized.csv", delimiter= ',', usecols=['timestamp_utc', 'looted_by__name', 'item_id', 'quantity'])
date_format = '%Y-%m-%dT%H:%M:%S.%f'
num_of_runs = 0


for index1,row1 in combined_df_3.iterrows():
    num_of_runs += 1
    try:
        row2 = combined_df_4.iloc[num_of_runs]
    except IndexError:
        break
    time_stamp_1 = datetime.strptime(row1['timestamp_utc'][:26], date_format)
    name_1 = row1['looted_by__name']
    item_id_1 = row1['item_id']
    quantity_1 = row1['quantity']
    
    time_stamp_2 = datetime.strptime(row2['timestamp_utc'][:26], date_format)
    name_2 = row2['looted_by__name']
    item_id_2 = row2['item_id']
    quantity_2 = row2['quantity']

    if name_1 != name_2 or item_id_1 != item_id_2 or quantity_1 != quantity_2:
        continue
    elif time_stamp_1 > time_stamp_2:
        date_diff_1 = abs(time_stamp_1 - time_stamp_2)
        date_diff_sec_1 = date_diff_1.total_seconds()
        if date_diff_sec_1 < 0.5:
            #donot add row 1 to df and add row 1 and row 2 to bugged item list
            bugged_item_df = bugged_item_df._append(row1 ,ignore_index=True)
            bugged_item_df = bugged_item_df._append(row2 ,ignore_index=True)
            bugged_item_2df = bugged_item_2df._append(row1,ignore_index=True)
            

    elif time_stamp_1 < time_stamp_2:
        date_diff_2 = abs(time_stamp_2 - time_stamp_1)
        date_diff_sec_2 = date_diff_2.total_seconds()
        if date_diff_sec_2 < 0.5:
            bugged_item_df = bugged_item_df._append(row1 ,ignore_index=True)
            bugged_item_df = bugged_item_df._append(row2 ,ignore_index=True)
            bugged_item_2df = bugged_item_2df._append(row1,ignore_index=True)
            #donot add row 1 to df and add row 1 and row 2 to bugged item list
            
    

bugged_item_df.to_csv("bugged.csv", index=False)
print('here')
clean_item_df = combined_df_3.merge(bugged_item_2df, on=['timestamp_utc', 'looted_by__name', 'item_id', 'quantity'], how='left', indicator=True).query('_merge == "left_only"').drop('_merge', axis=1)
clean_item_df.to_csv("clean.csv", index=False)

求助 #

目前耗时60秒，希望能进一步优化到30秒以内，欢迎分享方案。

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