一、为什么用 Pandas 分析日志

服务器日志通常以文本行形式持续写入，单日访问量动辄几十万到上千万行。直接用 grep、awk 处理虽然快，但面对多维度的统计需求（按状态码、按接口、按时间段、按客户端 IP 聚合）时，代码会迅速变得难以维护。

Pandas 的优势在于把日志转换成结构化的 DataFrame 之后，过滤、分组、聚合、排序都可以用几行代码完成，非常适合做异常请求的探索性分析。本文以 Nginx 的 access.log 为例，完整走一遍从原始日志到异常请求定位的流程。

二、准备日志数据

Nginx 默认的 combined 格式日志长这样：

192.168.1.10 - - [10/Nov/2026:14:23:01 +0800] "GET /api/order/list?page=1 HTTP/1.1" 200 5321 "https://example.com/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
10.0.0.23 - - [10/Nov/2026:14:23:02 +0800] "POST /api/login HTTP/1.1" 401 128 "-" "python-requests/2.31.0"
172.16.5.8 - - [10/Nov/2026:14:23:05 +0800] "GET /api/order/list?page=1 HTTP/1.1" 500 0 "-" "curl/8.4.0"

为了便于解析，建议在 Nginx 中自定义一个更规整的 log_format，把字段用固定分隔符隔开，例如使用 | 或制表符：

log_format main '$remote_addr|$remote_user|$time_local|$request|$status|$body_bytes_sent|$request_time|$http_referer|$http_user_agent|$upstream_response_time';
access_log /var/log/nginx/access.log main;

如果无法修改线上配置，也可以直接用正则解析默认格式，后文会给出两种方式的代码。

三、把日志读入 DataFrame

3.1 使用分隔符直接读取

当日志字段使用固定分隔符时，read_csv 是最省事的方式：

import pandas as pd
columns = [
"remote_addr", "remote_user", "time_local", "request",
"status", "body_bytes_sent", "request_time",
"http_referer", "http_user_agent", "upstream_time",
]
df = pd.read_csv(
"/var/log/nginx/access.log",
sep="|",
names=columns,
na_values=["-"],
keep_default_na=False,
dtype={"status": "int32", "body_bytes_sent": "int64"},
)
print(df.shape)
print(df.head())

这里把 - 统一识别为缺失值，并显式指定了数值列的类型，可以显著降低内存占用。

3.2 使用正则解析默认格式

如果日志是默认 combined 格式，可以用正则一次性提取字段：

import re
import pandas as pd
pattern = re.compile(
r'(?P<remote_addr>\S+) \S+ \S+ '
r'[(?P<time_local>[^]]+)] '
r'"(?P<request>[^"])" '
r'(?P<status>\d{3}) (?P<body_bytes_sent>\d+) '
r'"(?P<http_referer>[^"])" '
r'"(?P<http_user_agent>[^"]*)"'
)
rows = []
with open("/var/log/nginx/access.log", encoding="utf-8", errors="ignore") as f:
for line in f:
m = pattern.match(line)
if m:
rows.append(m.groupdict())
df = pd.DataFrame(rows)
df["status"] = df["status"].astype("int32")
df["body_bytes_sent"] = df["body_bytes_sent"].astype("int64")
print(df.shape)

逐行正则解析速度较慢，百万行级别建议先用 awk 或 sed 预处理成结构化文本，再交给 Pandas。

四、数据清洗与字段派生

原始字段还不能直接用于分析，需要做几件事：解析时间、拆分请求方法、路径和协议、提取 URL 参数。

import pandas as pd
1. 解析时间
df["time_local"] = pd.to_datetime(
df["time_local"], format="%d/%b/%Y:%H:%M:%S %z", errors="coerce"
)
df = df.dropna(subset=["time_local"])
2. 拆分 request 为 method / path / protocol
req = df["request"].str.split(" ", n=2, expand=True)
df["method"] = req[0]
df["path"] = req[1]
df["protocol"] = req[2]
3. 分离路径与查询参数
df["path_only"] = df["path"].str.split("?", n=1).str[0]
4. 派生时间维度
df["hour"] = df["time_local"].dt.floor("h")
df["minute"] = df["time_local"].dt.floor("min")
5. 数值字段兜底
df["request_time"] = pd.to_numeric(df["request_time"], errors="coerce")
df["upstream_time"] = pd.to_numeric(df["upstream_time"], errors="coerce")
print(df[["time_local", "method", "path_only", "status", "request_time"]].head())

清洗完成后，df 就具备了按时间、接口、状态码、耗时等多个维度分析的基础。

五、整体健康度概览

在定位异常之前，先看几个全局指标，建立对流量规模的直观认识。

total = len(df)
err_rate = (df["status"] >= 500).mean()
p95_rt = df["request_time"].quantile(0.95)
print(f"总请求数: {total:,}")
print(f"5xx 占比: {err_rate:.2%}")
print(f"P95 响应时间: {p95_rt:.3f}s")
状态码分布
print(df["status"].value_counts().sort_index())
请求方法分布
print(df["method"].value_counts())

如果 5xx 占比超过 1%，或者 P95 响应时间明显高于平时，就说明存在需要深入排查的异常。

六、定位异常请求的五个维度

6.1 按状态码定位服务端错误

5xx 是最直接的异常信号，先按接口聚合，找出错误集中的路径：

server_err = df[df["status"] >= 500]
top_err_path = (
    server_err.groupby("path_only")
    .agg(err_count=("status", "size"),
         avg_rt=("request_time", "mean"))
    .sort_values("err_count", ascending=False)
    .head(20)
)
print(top_err_path)

同时可以按分钟统计 5xx 数量，判断是持续性问题还是某个时间点的突发故障：

err_by_minute = (
    server_err.groupby("minute").size().sort_values(ascending=False).head(10)
)
print(err_by_minute)

6.2 按响应时间定位慢请求

响应时间异常往往先于错误出现，是更早期的预警信号。用分位数而不是平均值来判断，可以避免被少量极端值误导：

slow_threshold = df["request_time"].quantile(0.99)
slow = df[df["request_time"] >= slow_threshold]
slow_by_path = (
slow.groupby("path_only")
.agg(count=("request_time", "size"),
p99=("request_time", lambda s: s.quantile(0.99)),
max_rt=("request_time", "max"))
.sort_values("count", ascending=False)
.head(20)
)
print(slow_by_path)

如果某个接口的 P99 远高于其他接口，通常意味着该接口存在慢查询、外部依赖超时或数据量过大等问题。

6.3 按客户端 IP 定位异常来源

单个 IP 请求量异常高，可能是爬虫、压测流量或攻击行为：

ip_stats = (
    df.groupby("remote_addr")
    .agg(req_count=("status", "size"),
         err_count=("status", lambda s: (s >= 400).sum()),
         avg_rt=("request_time", "mean"),
         ua=("http_user_agent", "first"))
    .sort_values("req_count", ascending=False)
    .head(20)
)
ip_stats["err_ratio"] = ip_stats["err_count"] / ip_stats["req_count"]
print(ip_stats)

重点关注两类 IP：请求量极高但错误率也高的（可能是扫描器），以及请求量极高且集中在少数接口的（可能是爬虫）。

6.4 按时间窗口定位突发流量

把请求量按分钟聚合，可以直观看到流量尖峰：

qps = df.groupby("minute").size()
baseline = qps.median()
spikes = qps[qps > baseline * 3].sort_values(ascending=False)
print(f"基线 QPS: {baseline:.0f}")
print(spikes.head(10))

超过基线 3 倍的分钟通常对应活动推广、爬虫抓取或异常重试风暴，可以进一步下钻到该时间窗口内的 IP 和接口分布。

6.5 按 User-Agent 识别非正常客户端

ua_stats = (
    df.groupby("http_user_agent")
    .agg(req_count=("status", "size"),
         err_count=("status", lambda s: (s >= 400).sum()))
    .sort_values("req_count", ascending=False)
    .head(15)
)
print(ua_stats)

空 UA、curl、python-requests 等非浏览器 UA 出现大量请求时，需要结合业务判断是否为正常调用方。

七、组合条件筛选可疑请求

单一维度容易误判，把多个信号组合起来，可以更精准地锁定真正可疑的请求：

suspicious = df[
    (df["status"] >= 400)
    & (df["request_time"] > 1.0)
    & (df["path_only"].str.startswith("/api/"))
]
summary = (
suspicious.groupby(["remote_addr", "path_only"])
.agg(count=("status", "size"),
statuses=("status", lambda s: sorted(s.unique().tolist())),
avg_rt=("request_time", "mean"))
.sort_values("count", ascending=False)
.head(20)
)
print(summary)

这段代码筛出「访问 API 接口、返回 4xx/5xx、且耗时超过 1 秒」的请求，再按 IP 和接口聚合，通常能直接暴露出问题源头。

八、输出分析报告

把关键结论导出为文件，便于后续复盘和团队共享：

with pd.ExcelWriter("log_analysis.xlsx") as writer:
    df["status"].value_counts().to_frame("count").to_excel(
        writer, sheet_name="status_dist")
    top_err_path.to_excel(writer, sheet_name="top_error_path")
    slow_by_path.to_excel(writer, sheet_name="slow_path")
    ip_stats.to_excel(writer, sheet_name="top_ip")
    summary.to_excel(writer, sheet_name="suspicious")
print("分析报告已导出: log_analysis.xlsx")

九、性能优化建议

  • 只读需要的列：使用 usecols 参数，避免把 UA、referer 等长文本全部载入内存。
  • 指定 dtype：状态码用 int32，字节数用 int64，避免默认 object 类型浪费内存。
  • 分块读取：超大日志用 chunksize 分批处理，边读边聚合，避免一次性加载。
  • 优先用向量化操作：str.split、dt.floor、groupby.agg 都比 apply 逐行处理快得多。
  • 善用 category 类型：状态码、方法、路径等低基数列转为 category，可进一步压缩内存。
chunk_iter = pd.read_csv(
    "/var/log/nginx/access.log",
    sep="|",
    names=columns,
    usecols=["remote_addr", "time_local", "request", "status", "request_time"],
    dtype={"status": "int32"},
    chunksize=200_000,
)
parts = []
for chunk in chunk_iter:
chunk = chunk[chunk["status"] >= 400]
parts.append(chunk)
err_df = pd.concat(parts, ignore_index=True)
print(err_df.shape)

十、小结

用 Pandas 分析服务器日志的核心思路是：先把非结构化文本转成规整的 DataFrame，再通过状态码、响应时间、客户端 IP、时间窗口和 User-Agent 五个维度逐层下钻，最后用组合条件收敛到具体的可疑请求。

实际排查时建议按「全局概览 → 单维度异常 → 组合筛选 → 时间点下钻」的顺序推进，既能快速定位问题，也能避免被单一指标误导。对于持续增长的日志量，可以进一步结合分块读取和列裁剪，把分析脚本稳定运行在常规配置的机器上。

# eclipse # spring # intellij-idea # spring cloud # servlet
Logo
北京朝阳AI社区

更多推荐

cover

每周一道 Agent 面试题 - 现在市面上的 Embedding Model 怎么选择(6)

avatar 北京朝阳AI社区
cover

防拍照系统的工程分水岭

avatar 北京朝阳AI社区
cover

FDE 的圈子不在“FDE”这三个字里

avatar 北京朝阳AI社区