如何高效匹配两个pandas DataFrame中符合IP及时间条件的对应行

核心优化思路 #

逐行遍历的时间复杂度为O(M*N)（M为flows行数，N为attacks行数），在数据量大时效率极低。我们可以通过先缩小匹配范围，再做条件过滤的思路，把复杂度降到O(M log M + N log N)，核心逻辑是：

1. 你的IP匹配条件本质是「攻击的源目的IP集合与流的源目的IP集合完全相等，不区分顺序」，所以可以提前生成有序IP对作为关联键，先过滤掉所有IP不匹配的行
2. 仅对IP匹配的少量行做时间范围校验，全程用pandas向量化操作实现，避免Python层循环

实现代码 #

方案1：等值merge+条件过滤（适合10万级以内数据，代码最简洁） #

import pandas as pd
import numpy as np

# 生成有序IP对作为关联键，用numpy向量化操作比apply快数倍
flows_ip_arr = np.sort(flows[['sourceIPAddress', 'destinationIPAddress']].values, axis=1)
flows['ip_pair'] = list(map(tuple, flows_ip_arr))

attacks_ip_arr = np.sort(attacks[['srcIP', 'dstIP']].values, axis=1)
attacks['ip_pair'] = list(map(tuple, attacks_ip_arr))

# 先通过IP对做等值连接，过滤掉99%以上不相关的行
merged = pd.merge(
    flows.reset_index(), 
    attacks[['ip_pair', 'datetime']], 
    on='ip_pair', 
    how='left'
)

# 过滤满足时间范围的匹配项
valid_matches = merged[
    (merged['datetime'] >= merged['flowStartMicroseconds']) 
    & (merged['datetime'] <= merged['flowEndMicroseconds'])
]

# 给flows新增匹配标记列
flows['is_attack_related'] = flows.index.isin(valid_matches['index'].unique())

方案2：分组+merge_asof范围匹配（适合百万级以上数据，内存占用更低） #

# 按IP对和时间排序，为merge_asof做准备
flows_sorted = flows.sort_values(['ip_pair', 'flowStartMicroseconds']).reset_index()
attacks_sorted = attacks.sort_values(['ip_pair', 'datetime'])[['ip_pair', 'datetime']]

matched_indexes = []
# 按IP对分组匹配，避免大表笛卡尔积
for ip_pair, flow_group in flows_sorted.groupby('ip_pair'):
    attack_group = attacks_sorted[attacks_sorted['ip_pair'] == ip_pair]
    if len(attack_group) == 0:
        continue
    # 用merge_asof做时间范围匹配，性能远高于普通条件过滤
    group_match = pd.merge_asof(
        attack_group,
        flow_group,
        left_on='datetime',
        right_on='flowStartMicroseconds',
        by='ip_pair'
    )
    valid_group = group_match[group_match['datetime'] <= group_match['flowEndMicroseconds']]
    matched_indexes.extend(valid_group['index'].tolist())

flows['is_attack_related'] = flows.index.isin(matched_indexes)

超大量级适配建议 #

如果两个表的行数达到千万级以上，建议转用PySpark的范围连接功能，或者用Dask做分布式处理，核心逻辑和上述方案一致，只是把pandas操作换成对应框架的API即可。

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