基于Pandas按IP和时间间隔统计正则匹配日志次数的方案求助

问题背景 #

现有日志数据结构（字典列表）：

list1 =[
{'ip': '11.22.33.44', 'timestamp': 1665480231699, 'message': '{"body": "Idle time larger than time period. retry:0"}', 'ingestionTime': 1665480263198},
{'ip': '11.22.33.42', 'timestamp': 1665480231698, 'message': '{"body": "Idle time larger than time period. retry:5"}', 'ingestionTime': 1665480263198}, 
{'ip': '11.22.33.44', 'timestamp': 1665480231698, 'message': '{"body": "Idle time larger than time period. retry:0"}', 'ingestionTime': 1665480263198}
]

需要匹配的白名单规则列表（含正则表达式）：

whitelist_metadata = [
  {
    'LogLevel': 'WARNING',
    'SpecificVersion': 'None',
    'TimeInterval(Min)': 1,
    'MetricMsg': 'DDR: XXXX count got lost',
    'AllowedOccurrenceInTimeInterval': 0   # 0表示始终允许出现
  },
  {
    'LogLevel': 'WARNING',
    'SpecificVersion': 'None',
    'TimeInterval(Min)': 1,
    'MetricMsg': r'Idle time larger than .* time\. retry: \d+',     # 正则表达式
    'AllowedOccurrenceInTimeInterval': 5  # 1分钟内最多允许出现5次
  }
]

需求：按单个IP维度，检查每条日志消息是否匹配白名单中的正则，统计指定时间间隔内的出现次数：

• 若规则允许次数为0：直接保留该记录
• 若次数超过规则阈值：保留记录；否则移除该记录

此前尝试使用Pandas rolling("5m") 时触发报错：ValueError('window must be an integer',)，需要高性能的实现方案。

解决方案 #

步骤1：数据预处理与格式转换 #

将数据转为DataFrame，转换时间戳格式并解析日志消息中的正文内容：

import pandas as pd
import json
import re

# 加载数据
list1 =[
{'ip': '11.22.33.44', 'timestamp': 1665480231699, 'message': '{"body": "Idle time larger than time period. retry:0"}', 'ingestionTime': 1665480263198},
{'ip': '11.22.33.42', 'timestamp': 1665480231698, 'message': '{"body": "Idle time larger than time period. retry:5"}', 'ingestionTime': 1665480263198}, 
{'ip': '11.22.33.44', 'timestamp': 1665480231698, 'message': '{"body": "Idle time larger than time period. retry:0"}', 'ingestionTime': 1665480263198}
]

# 转为DataFrame并处理时间戳
df = pd.DataFrame(list1)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')

# 解析message中的body字段，方便后续正则匹配
df['body'] = df['message'].apply(lambda x: json.loads(x)['body'])

步骤2：匹配白名单规则 #

遍历白名单规则，为每条日志标记匹配的规则及对应的阈值：

# 初始化匹配标记列
df['matched_rule'] = None
df['allowed_count'] = None
df['time_interval'] = None

for rule in whitelist_metadata:
    pattern = re.compile(rule['MetricMsg'])
    # 匹配当前规则的日志行
    match_mask = df['body'].str.contains(pattern)
    # 填充匹配规则的相关参数
    df.loc[match_mask, 'matched_rule'] = rule['MetricMsg']
    df.loc[match_mask, 'allowed_count'] = rule['AllowedOccurrenceInTimeInterval']
    df.loc[match_mask, 'time_interval'] = pd.Timedelta(minutes=rule['TimeInterval(Min)'])

步骤3：按IP+规则分组，统计时间窗口内的出现次数 #

使用时间窗口滚动计数的核心要求：必须将timestamp设置为分组后的索引，否则会触发窗口类型错误。

# 拆分有匹配规则和无匹配规则的日志
matched_df = df[df['matched_rule'].notna()]
unmatched_df = df[df['matched_rule'].isna()]

# 按IP和匹配规则分组，设置timestamp为索引并排序
grouped = matched_df.set_index('timestamp').sort_index().groupby(['ip', 'matched_rule'])

# 定义滚动计数逻辑
def calculate_rolling_count(group):
    window_duration = group['time_interval'].iloc[0]
    # 统计当前时间窗口内的日志出现次数
    group['rolling_count'] = group.rolling(window_duration)['body'].count()
    return group

# 应用滚动计数到每个分组
processed_grouped = grouped.apply(calculate_rolling_count).reset_index()

步骤4：根据阈值过滤记录 #

根据规则的允许次数过滤日志：

# 生成保留规则的掩码：允许次数为0 或 滚动计数超过阈值
keep_mask = (processed_grouped['allowed_count'] == 0) | (processed_grouped['rolling_count'] > processed_grouped['allowed_count'])
filtered_matched = processed_grouped[keep_mask]

# 合并过滤后的匹配日志和未匹配日志
final_df = pd.concat([filtered_matched, unmatched_df], ignore_index=True)

# 可选：恢复原数据的列结构
final_df = final_df[list1[0].keys()]

关键说明 #

• 使用rolling时间窗口时，必须确保分组后的DataFrame索引为datetime类型，否则会触发window must be an integer错误；
• 按IP+规则分组可避免不同IP、不同规则的日志互相干扰；
• 基于Pandas向量化操作实现，处理大规模日志数据时性能远优于纯循环遍历。

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