Python 数据分析实战：用 Pandas 清洗百万行日志数据

# Python 数据分析实战：用 Pandas 清洗百万行日志数据 ## 场景 我们有一份 200 万行的 Nginx 访问日志，需要分析 PV/UV 趋势、TOP 路径、异常状态码分布。 ## 第一步：读取与解析 ```python import pandas as pd import re pattern = re.compile( r'(?P[\d.]+) - - \[(?P[^\]]+)\] ' r'"(?P\w+) (?P[^ ]+) [^"]*" ' r'(?P\d+) (?P\d+)' ) def parse_line(line): m = pattern.match(line) if m: return m.groupdict() return None # 分块读取，避免内存溢出 chunks = [] with open('access.log', 'r') as f: batch = [] for line in f: parsed = parse_line(line) if parsed: batch.append(parsed) if len(batch) >= 100000: chunks.append(pd.DataFrame(batch)) batch = [] if batch: chunks.append(pd.DataFrame(batch)) df = pd.concat(chunks, ignore_index=True) ``` ## 第二步：数据类型转换与清洗 ```python df['time'] = pd.to_datetime( df['time'], format='%d/%b/%Y:%H:%M:%S %z' ) df['status'] = df['status'].astype(int) df['size'] = df['size'].astype(int) df['date'] = df['time'].dt.date df['hour'] = df['time'].dt.hour ``` ## 第三步：分析 ```python # PV/UV 日趋势 daily = df.groupby('date').agg( pv=('ip', 'count'), uv=('ip', 'nunique') ).reset_index() # TOP 10 路径 top_paths = df['path'].value_counts().head(10) # 异常请求（4xx/5xx） errors = df[df['status'] >= 400] error_rate = len(errors) / len(df) * 100 print('错误率: {:.2f}%'.format(error_rate)) ``` ## 优化技巧 1. 使用 `dtype` 参数预定义列类型减少内存占用 2. 用 `pd.Categorical` 转换低基数列（如 method） 3. 大文件用 `dask` 替代 `pandas` 实现并行处理