4、数据采集规范：数据源确认、采集频率、采集工具选择、数据清洗与预处理

数据采集这事儿，看着简单，其实坑最多。我见过太多团队，花大价钱搭了数据平台，结果采回来的数据全是垃圾——说白了，就是采集阶段没管好。

这一章，咱们就把数据采集的四个关键环节掰开揉碎讲清楚。你按这个流程走，至少能保证采回来的数据是“能用的”。

4.1 数据源确认：别让源头坑了你

数据源确认，说白了就是搞清楚“数据从哪来、靠不靠谱”。我刚开始做数据项目时，接过一个需求，对方说“数据都在ERP里”，结果一查，ERP里存的是手工录入的Excel导入数据，连字段名都对不上。嗯，从那以后，我养成了一个习惯——先验证数据源，再谈采集方案。

数据源确认要回答三个问题：

1. 数据源是否权威？ 是系统自动生成的，还是人工录入的？人工录入的数据，出错概率高得多。
2. 数据源是否稳定？ 接口会不会经常断？数据库会不会半夜重启？
3. 数据源是否完整？ 有没有缺失字段？历史数据能不能回溯？

4.2 采集频率：不是越快越好

很多人觉得，采集频率越高越好。其实不然。你想想看，一个库存表每天只更新一次，你每5分钟去拉一次，除了浪费带宽和存储，没有任何意义。

采集频率的设定，我一般按这个逻辑来：

数据变化特征	推荐采集频率	典型场景
实时变化（秒级）	流式采集（Kafka/Pulsar）	交易流水、传感器数据
频繁变化（分钟级）	每5-15分钟批量拉取	订单状态、库存水位
日常更新（小时/天级）	每日定时批量同步	客户信息、产品目录
低频更新（周/月级）	按需触发或周度同步	财务报表、年度预算

我曾经踩过一个坑：把某个日更报表的采集频率设成了每分钟一次。结果一个月后，存储成本翻了3倍，下游数仓被撑爆了。后来改成每天凌晨2点采集一次，什么问题都没有。

4.3 采集工具选择：合适比流行重要

市面上的采集工具多得很，但别盲目追新。我见过有人用Flink去采集一个每天只有几百条记录的CRM表——杀鸡用牛刀，运维成本还高。

工具选择，我一般看三个维度：

• 数据量级： 小数据量（<100万条/天）用Sqoop或DataX就够了；大数据量（>1000万条/天）才考虑Flink、Spark Streaming。
• 数据格式： 结构化数据用JDBC/ODBC；半结构化（JSON/XML）用Logstash或Filebeat；非结构化（图片/文档）用对象存储+事件触发。
• 团队能力： 团队熟悉Java就用Flink，熟悉Python就用Airbyte或自定义脚本。别为了“技术先进”选一个没人能维护的工具。

4.4 数据清洗与预处理：脏数据必须在这里拦住

数据清洗，是采集环节里最脏最累的活，但也是最有价值的。我常说一句话：清洗做得好，分析没烦恼。

清洗流程我一般分四步走：

1. 去重： 同一笔数据被采集了多次？用主键+时间戳去重。注意，有些场景下“重复”是合法的（比如重试机制），要区分对待。
2. 空值处理： 空值不能直接扔掉。先判断是“数据缺失”还是“数据不存在”。缺失的可以填充默认值或插值；不存在的要保留空标记。
3. 格式统一： 日期格式、数字精度、编码方式必须统一。我遇到过最离谱的——同一个“2024-01-01”，有的系统存成“2024/01/01”，有的存成“01-01-2024”。
4. 异常值检测： 用统计方法（3σ、IQR）或规则引擎，把明显不合理的数据标记出来。比如年龄500岁、订单金额负数——这些数据不能直接丢弃，但必须打标。

下面是一个简单的清洗脚本示例，我经常用它做快速预处理：

import pandas as pd
import numpy as np

def clean_raw_data(df):
    # 1. 去重：基于主键去重，保留最新记录
    df = df.drop_duplicates(subset=['order_id'], keep='last')
    
    # 2. 空值处理：数值字段填充中位数，分类字段填充'unknown'
    for col in df.select_dtypes(include=[np.number]).columns:
        df[col].fillna(df[col].median(), inplace=True)
    for col in df.select_dtypes(include=['object']).columns:
        df[col].fillna('unknown', inplace=True)
    
    # 3. 格式统一：日期统一为YYYY-MM-DD
    df['create_date'] = pd.to_datetime(df['create_date']).dt.strftime('%Y-%m-%d')
    
    # 4. 异常值标记：金额为负或超过3σ的标记为异常
    mean_val = df['amount'].mean()
    std_val = df['amount'].std()
    df['amount_anomaly'] = ((df['amount'] < 0) | 
                            (np.abs(df['amount'] - mean_val) > 3 * std_val))
    
    return df

4.5 数据质量标记：给每一条数据一个“健康证”

清洗完的数据，不能直接扔给下游。我建议给每条数据加一个质量标记字段，比如：

• quality = 'good'：数据完整、无异常，可直接使用
• quality = 'warning'：数据有轻微问题（如空值被填充），需谨慎使用
• quality = 'bad'：数据存在严重异常，不建议用于分析

这样下游的分析师拿到数据后，一眼就能知道哪些数据是“干净的”，哪些需要再确认。说白了，这就是给数据发“健康证”。