日志清洗与预处理：数据去重、字段提取、时间戳标准化、脏数据过滤

说实话，很多做搜索日志分析的朋友，一上来就急着做聚合、算指标。我见过太多团队，花了两周搭好看板，结果数据全是错的。为什么？因为日志本身就没洗干净。

日志清洗，说白了就是给数据「洗澡」。你从线上收上来的日志，什么脏东西都有——重复的、格式乱的、字段缺失的、时间戳乱七八糟的。不洗直接分析，结果就是垃圾进垃圾出。

我个人习惯，把清洗分成四个步骤：去重、字段提取、时间标准化、脏数据过滤。每一步都有坑，咱们一个一个说。

1. 数据去重：别让同一条日志算两次

日志重复，是最常见的问题。网络重传、客户端重试、服务端幂等没做好，都会导致同一条行为被记录多次。

我在项目中遇到过，某次大促活动，PV 指标直接翻了三倍。查了半天，发现是客户端在弱网环境下自动重试了三次请求，日志就收了三条一模一样的。

去重怎么做？核心是找到唯一标识。对于搜索日志，我一般用 session_id + timestamp + query 拼接成一个指纹，然后做 hash 去重。

去重策略优先级：
  • 业务主键去重：比如 request_id、log_id，最可靠
  • 多字段组合去重：没有唯一 ID 时，用 session_id + 时间戳 + 事件类型
  • 滑动窗口去重：对于流式处理，用时间窗口内的近似去重（比如布隆过滤器）
# 伪代码示例：基于 request_id 的去重
def dedup_logs(logs):
    seen = set()
    clean = []
    for log in logs:
        fingerprint = log['request_id']
        if fingerprint not in seen:
            seen.add(fingerprint)
            clean.append(log)
    return clean
我的经验：如果日志量特别大（每天几十亿条），别用内存里的 set 去重。用 Redis 的 set 或者 HyperLogLog，能省 90% 的内存。

3. 字段提取：从原始日志里捞出有用的东西

原始日志通常是一大段文本，或者一个 JSON 字符串。你需要从中提取出关键字段，比如：搜索词、用户 ID、点击的商品 ID、页面位置、耗时等。

嗯，这里要注意。很多新手喜欢用正则表达式硬匹配，结果性能差、还容易出错。我建议优先用结构化的解析方式。

举个例子，如果你的日志是 JSON 格式，直接用 JSON 解析器提取字段。如果是自定义格式，比如用竖线分隔的，那就用 split 方法。

# 结构化日志解析示例
import json

def extract_fields(raw_log):
    try:
        parsed = json.loads(raw_log)
        return {
            'user_id': parsed.get('uid'),
            'query': parsed.get('q'),
            'click_item': parsed.get('item_id'),
            'position': parsed.get('pos'),
            'duration_ms': parsed.get('dur')
        }
    except json.JSONDecodeError:
        return None  # 脏数据，后续过滤
避坑指南：我曾经遇到过一个坑——日志里某个字段的值包含了换行符，导致一行日志被拆成了两行。解析的时候一定要考虑字段值内部的特殊字符。

4. 时间戳标准化：统一时区、统一格式

搜索日志的时间戳，可能是最乱的字段。有的用 Unix 毫秒时间戳，有的用 ISO 8601 字符串，有的用「2024-01-15 14:30:00」这种格式。更麻烦的是，不同服务器可能处于不同时区。

我建议，所有时间戳统一转为 Unix 毫秒时间戳（或者纳秒，取决于你的精度需求）。为什么？因为做时间范围查询、计算耗时、做时序分析，数值型时间戳最方便。

原始格式 示例 标准化后
Unix 秒 1705300000 1705300000000
ISO 8601 2024-01-15T06:30:00Z 1705300200000
自定义格式 2024/01/15 14:30:00 1705300200000 
# 时间戳标准化示例
from datetime import datetime

def normalize_timestamp(ts_str):
    # 尝试多种格式
    formats = [
        '%Y-%m-%dT%H:%M:%SZ',
        '%Y-%m-%d %H:%M:%S',
        '%Y/%m/%d %H:%M:%S'
    ]
    for fmt in formats:
        try:
            dt = datetime.strptime(ts_str, fmt)
            return int(dt.timestamp() * 1000)
        except ValueError:
            continue
    return None  # 无法解析，视为脏数据
一个小技巧：如果日志量特别大，别在清洗阶段做时区转换。统一先存成 UTC 毫秒时间戳，等查询展示的时候再转本地时间。这样能避免很多时区相关的 bug。

5. 脏数据过滤：把垃圾挡在门外

脏数据，就是那些明显不合理、或者格式错误的数据。比如：搜索词为空、用户 ID 为负数、耗时字段是负数、点击位置超出页面范围等等。

你想想看，如果一条日志里搜索词是空字符串，那这条日志对分析搜索行为有什么意义？没有。直接丢掉就好。

我一般会建一个「脏数据规则表」，把常见的过滤规则列出来。这样既清晰，又方便后续维护。

# 脏数据过滤规则示例
FILTER_RULES = [
    ('query 为空', lambda log: len(log.get('query', '').strip()) > 0),
    ('duration 为负数', lambda log: log.get('duration_ms', 0) >= 0),
    ('user_id 非法', lambda log: log.get('user_id', 0) > 0),
    ('position 越界', lambda log: 0 <= log.get('position', -1) <= 100),
]

def filter_dirty(log):
    for rule_name, rule_func in FILTER_RULES:
        if not rule_func(log):
            return False  # 命中规则，过滤掉
    return True
注意：过滤脏数据的时候，一定要记录被过滤掉的数据量和原因。我曾经因为过滤规则太严，把 30% 的正常日志都丢了，查了两天才发现是规则写错了。建议把过滤日志单独存一份，方便复盘。

6. 清洗流程的整体设计

上面四个步骤，不是串行执行的。我建议的流程是：

  1. 先做字段提取：把原始日志解析成结构化数据
  2. 再做脏数据过滤：把明显有问题的数据先扔掉
  3. 然后做时间戳标准化：统一时间格式
  4. 最后做数据去重：去掉重复记录

为什么去重要放最后？因为如果先做去重，后面过滤脏数据的时候，可能会把一些「看起来重复但其实是不同用户」的数据误删。先过滤、再去重，更安全。

总结一下：
  • 去重：找唯一标识，用 hash 或布隆过滤器
  • 字段提取：优先用结构化解析，少用正则
  • 时间戳标准化：统一转 Unix 毫秒，存 UTC
  • 脏数据过滤：建规则表，记录过滤日志

日志清洗这件事，看起来简单，但做不好后面全白搭。我见过太多团队，花 80% 的时间搭分析平台，只花 20% 的时间做清洗。结果平台搭好了，数据不敢用。你想想看，是不是这个道理？

下一章，咱们聊聊清洗完的数据怎么存储和索引。到时候我会分享一些我在 Elasticsearch 和 ClickHouse 之间做选择的经验。

← 上一章 📖 返回目录 下一章 →