使用Pandas筛选CSV数据并关联前置特定条件行的方案问询

需求 #

现有两个仅Error_2列值不同的CSV文件，需完成以下操作：

1. 筛选出Error_2='SB_Data_Expiration'且Error_3='Loss_of_Data'的行，保留Date/Time、Error_1、Error_2、Error_3列；
2. 为筛选出的行添加Previous_Date/Time列，取值为该行时间之前最近的Error_2='Loss_of_Data'对应的Date/Time值。

期望输出示例 #

• 样本1输出：

Date/Time                Error_1            Error_2                 Error_3         Previous_Date/Time 
2023/01/01 17:11:04      APP_ERR         SB_Data_Expiration         Loss_of_Data    2023/01/01 17:10:44

• 样本2输出：

Date/Time                Error_1            Error_2                 Error_3         Previous_Date/Time 
2023/01/01 17:11:04      APP_ERR         SB_Data_Expiration         Loss_of_Data    2023/01/01 16:57:08

实现代码 #

步骤1：读取并预处理数据 #

确保时间列能正确排序和匹配：

import pandas as pd

# 读取CSV文件
df = pd.read_csv('your_csv_file.csv')

# 转换时间列为datetime类型，保证时间运算正常
df['Date/Time'] = pd.to_datetime(df['Date/Time'], format='%Y/%m/%d %H:%M:%S')

# 按时间排序（关键，否则匹配逻辑会出错）
df = df.sort_values('Date/Time').reset_index(drop=True)

步骤2：标记关键行 #

区分需要保留的目标行和用于匹配的参考行：

# 标记目标行
df['is_target'] = (df['Error_2'] == 'SB_Data_Expiration') & (df['Error_3'] == 'Loss_of_Data')

# 提取所有Error_2为Loss_of_Data的时间序列，作为匹配数据源
ref_time_series = df.loc[df['Error_2'] == 'Loss_of_Data', 'Date/Time']

步骤3：匹配最近前置时间 #

用二分查找高效定位每个目标行对应的最近前置时间：

def find_previous_time(row):
    if not row['is_target']:
        return pd.NaT
    # 找到第一个大于当前时间的参考时间索引，取前一个即为最近的前置时间
    idx = ref_time_series.searchsorted(row['Date/Time'], side='right') - 1
    return ref_time_series.iloc[idx] if idx >= 0 else pd.NaT

# 生成Previous_Date/Time列
df['Previous_Date/Time'] = df.apply(find_previous_time, axis=1)

步骤4：筛选并格式化结果 #

得到最终符合要求的输出：

# 筛选目标行，保留指定列
final_result = df.loc[df['is_target'], [
    'Date/Time', 'Error_1', 'Error_2', 'Error_3', 'Previous_Date/Time'
]]

# 将时间列转换回原字符串格式
final_result['Date/Time'] = final_result['Date/Time'].dt.strftime('%Y/%m/%d %H:%M:%S')
final_result['Previous_Date/Time'] = final_result['Previous_Date/Time'].dt.strftime('%Y/%m/%d %H:%M:%S')

# 打印结果或保存到CSV
print(final_result.to_string(index=False))
# final_result.to_csv('output.csv', index=False)

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