用 Python 归并多源时间线
多源时间线归并的难点不在排序,而在字段与时间语义的统一:各来源的时间列名不同、时区不同、精度也不同,直接拼接会得到一条错乱的时间轴。用 pandas 分三步处理更稳:读取时把时间列解析成时间类型、统一换算到 UTC、按时间排序后写回 CSV,作为后续分析唯一口径的输入。
基本信息
| 核心步骤 | read_csv 解析时间 → 统一 UTC → 拼接 → 排序 → 写回 CSV |
|---|---|
| 时区处理 | pd.to_datetime(..., utc=True) 或转换后再 tz_convert |
| 必备字段 | 时间、来源、对象、动作描述、原始文件与行号 |
| 列名对齐 | 拼接前把各源列名统一,避免同名字段语义不同 |
| 输出要求 | 保留来源列,任何一行都能回指原始日志 |
典型使用场景
- 把 Linux 与 Windows 两侧日志合并成统一时间线
- 处理同批数据中带时区偏移与不带偏移混用的情况
- 交付前生成固定格式的时间线 CSV,便于接收方复核
- 统计各来源的事件数量与时间分布,查看证据覆盖面
- 为后续的去重与自洽性检查准备干净输入
归并示例
归并示例
import pandas as pd
a = pd.read_csv('linux.csv', parse_dates=['ts']) # 读取并解析时间列
b = pd.read_csv('win.csv', parse_dates=['ts'])
tl = pd.concat([a, b], ignore_index=True) # 拼接,缺失列会被补齐为空
tl['ts'] = pd.to_datetime(tl['ts'], utc=True) # 统一换算到 UTC
tl = tl.dropna(subset=['ts']).sort_values('ts') # 丢掉解析失败的行再排序
tl.to_csv('merged.csv', index=False) # 写回统一口径的数据
print(tl['source'].value_counts()) # 核对各来源条数与原始文件是否相符归并检查项
| 检查项 | 方法 | 不合格表现 |
|---|---|---|
| 时间解析 | 检查是否出现 NaT | 解析失败被静默变成空值 |
| 时区统一 | 确认全部为 UTC 或全部带偏移 | 混用导致排序整体错乱 |
| 列名一致 | 拼接前比对两边列集合 | 同名不同义造成字段错读 |
| 来源可追溯 | 每行都有来源与原始文件标识 | 结论无法回指原始日志 |
拼接后先核对各来源的行数与原始文件是否吻合,数量对不上通常是解析或去重环节出了问题,先查清来源再往下做时间线,否则后面的分析都建立在残缺数据上。
不要用字符串排序代替时间排序:在时区不统一时字符串顺序会给出看似合理但错误的时间线,这类错误在报告里很难被发现。务必用时间类型字段排序,并抽查首尾各三行与原始日志核对。