3、数据采集与预处理：数据源接入（日志、数据库、消息队列）、数据清洗、缺失值处理、异常值检测

好，咱们直接进入正题。数据采集与预处理，说白了就是「把脏活累活干漂亮」。很多同学觉得这步没啥技术含量，上来就调模型、跑仿真。我刚开始也这么想，结果被真实数据狠狠教育了一回——数据源没接对，后面全白干。

今天我就把这块的实战经验掰开揉碎讲清楚。你想想看，仿真数据回放的核心是什么？是「还原真实场景」。如果源头数据就歪了，那回放出来的结果就是自欺欺人。

3.1 数据源接入：三种主流方式

数据源接入，我习惯把它分成三类：日志、数据库、消息队列。每种都有各自的脾气，得顺着来。

3.1.1 日志接入

日志是最常见的数据源。线上系统每时每刻都在吐日志，格式五花八门。我个人习惯用 Filebeat + Logstash 这套组合拳。

# Filebeat 配置示例（简化版）
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  multiline.pattern: '^\d{4}-\d{2}-\d{2}'
  multiline.negate: true
  multiline.match: after

output.logstash:
  hosts: ["logstash-server:5044"]

这里有个坑。日志里经常有多行消息（比如 Java 异常堆栈），如果不做 multiline 处理，一条异常会被拆成几十条记录。我曾经因为这个排查了整整一个下午，最后发现是配置漏了。

3.1.2 数据库接入

数据库接入，我建议用 CDC（Change Data Capture） 方案。为什么？因为全量同步太慢了，而且对业务库有压力。

常用的工具有 Debezium、Canal、Maxwell。我个人偏爱 Debezium，它基于 Kafka Connect，生态好，支持 MySQL、PostgreSQL、MongoDB 等多种数据库。

# Debezium MySQL 连接器配置（关键部分）
{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "debezium_pwd",
    "database.server.id": "184054",
    "database.server.name": "myapp",
    "database.include.list": "order_db",
    "table.include.list": "order_db.t_order, order_db.t_order_item",
    "snapshot.mode": "initial"
  }
}

嗯，这里要注意 snapshot.mode 参数。第一次启动时，Debezium 会做一次全量快照，之后才增量监听 binlog。如果你的表数据量很大（比如上亿条），快照阶段可能会很慢。我建议先做一次手动全量导出，再启动 CDC。

3.1.3 消息队列接入

消息队列接入，最常见的是 Kafka。仿真数据回放场景下，Kafka 几乎是标配。为什么？因为它能保存历史数据，支持回溯消费。

我一般这样配置消费者：

# Kafka 消费者配置（Python 示例）
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'order_topic',
    bootstrap_servers=['kafka1:9092', 'kafka2:9092'],
    auto_offset_reset='earliest',  # 从最早的消息开始消费
    enable_auto_commit=False,      # 手动提交偏移量
    group_id='data_replay_group',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

# 逐条消费，用于回放
for message in consumer:
    process_message(message.value)
    # 模拟真实时间间隔
    time.sleep(original_interval)

这里 auto_offset_reset='earliest' 很关键。仿真回放需要从头开始，而不是从最新消息开始。我见过有人忘了配这个，结果回放的数据全是最近几分钟的，完全失去了仿真意义。

3.2 数据清洗：把脏数据筛出去

数据清洗，说白了就是「去伪存真」。真实数据里什么妖魔鬼怪都有：字段错位、编码乱码、重复记录、格式不统一……

我总结了一套清洗流程，按优先级排序：

1. 格式清洗：统一日期格式、数字格式、编码格式
2. 去重清洗：基于业务主键去重，保留最新或最完整的一条
3. 逻辑清洗：检查字段间的逻辑关系（比如下单时间不能晚于支付时间）
4. 范围清洗：剔除超出合理范围的值（比如年龄 200 岁）

# 数据清洗示例（Python）
def clean_order_data(df):
    # 1. 格式清洗：统一日期格式
    df['order_time'] = pd.to_datetime(df['order_time'], errors='coerce')
    
    # 2. 去重清洗：按订单号去重，保留最后一条
    df = df.drop_duplicates(subset=['order_id'], keep='last')
    
    # 3. 逻辑清洗：支付时间不能早于下单时间
    invalid_mask = df['pay_time'] < df['order_time']
    df = df[~invalid_mask]
    
    # 4. 范围清洗：金额不能为负数
    df = df[df['amount'] >= 0]
    
    return df

3.3 缺失值处理：别乱填，先问为什么

缺失值处理，很多人上来就填均值、填中位数。我不建议这么干。为什么？因为缺失本身可能是一种信号。

我一般分三步走：

缺失类型	常见原因	处理方式
完全随机缺失	系统故障、网络抖动	删除该行，或用均值/中位数填充
随机缺失	用户未填写、字段可选	用模型预测填充，或用「未知」标记
非随机缺失	系统设计缺陷、特定条件下不采集	必须查明原因，不能随意填充

举个例子。用户画像数据里，「收入」字段经常缺失。如果你直接填均值，那低收入和高收入用户都会被拉到平均水平，仿真出来的用户行为就失真了。

我建议的做法是：

# 缺失值处理策略
def handle_missing(df, column, strategy='mark'):
    if strategy == 'mark':
        # 用特殊值标记，保留缺失信息
        df[column] = df[column].fillna(-999)
    elif strategy == 'model':
        # 用其他字段预测缺失值
        from sklearn.ensemble import RandomForestRegressor
        # ... 训练模型填充
    elif strategy == 'drop':
        # 缺失比例低时直接删除
        df = df.dropna(subset=[column])
    return df

3.4 异常值检测：别一刀切，要分场景

异常值检测，我把它分成两类：统计异常和业务异常。统计异常用算法找，业务异常靠规则筛。

3.4.1 统计异常检测

常用的方法有：

• 3σ 原则：适用于正态分布的数据
• 箱线图（IQR）：适用于偏态分布，更稳健
• Z-Score：标准化后看偏离程度
• 孤立森林：适用于高维数据

# 箱线图法检测异常值
def detect_outliers_iqr(df, column):
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    
    outliers = df[(df[column] < lower_bound) | (df[column] > upper_bound)]
    return outliers

3.4.2 业务异常检测

统计方法只能告诉你「这个值不太常见」，但到底是不是异常，还得看业务。比如：

• 凌晨 3 点的下单量突然暴增——可能是促销活动，也可能是爬虫攻击
• 用户单笔支付 100 万——可能是大客户，也可能是洗钱
• 接口响应时间 10 秒——可能是网络抖动，也可能是真的慢

我个人的处理原则是：先标记，再判断，最后处理。不要一上来就删除异常值，先看看它们是不是「有价值的异常」。

3.5 本章知识体系总览

说了这么多，我画了一张图帮你梳理整体流程。从数据源接入，到清洗、缺失值处理、异常值检测，每一步都有对应的策略和工具。

这张图把整个流程串起来了。你从左边选数据源，中间做清洗和处理，最后得到干净可用的数据。每一步都有对应的工具和方法，照着走基本不会出大问题。

可可精品资料库