数据采集质量：完整性、去重、延迟与异常处理

数据采集是整个链路的第一道关卡。我见过太多团队把精力都花在后面的分析模型上，结果源头数据一塌糊涂——缺字段、重复行、延迟到达、格式错乱。说白了，后面再好的算法也救不了垃圾数据。

这一章我们聚焦四个核心问题：完整性校验、去重策略、延迟控制、异常标记与重试。每个问题我都踩过坑，咱们一个一个说。

4.1 数据完整性校验

完整性校验，就是确认你收到的数据「该有的都有」。我习惯把它分成三个层次：

• 字段级：必填字段是否为空？格式对不对？
• 记录级：一条记录是否完整到达？有没有被截断？
• 批次级：这一批数据该有1000条，实际到了多少？

我在项目中常用的一种做法是「哈希链校验」。每条记录计算一个哈希值，批次结束时再算一个总哈希。如果上下游哈希不一致，说明数据在传输过程中被篡改或丢失了。

// 伪代码示例：批次完整性校验
class BatchValidator {
    private List<String> recordHashes = new ArrayList<>();
    
    public void addRecord(String rawData) {
        String hash = md5(rawData);
        recordHashes.add(hash);
    }
    
    public String computeBatchHash() {
        // 将所有记录的哈希拼接后再哈希
        String concatenated = String.join("", recordHashes);
        return sha256(concatenated);
    }
    
    public boolean verify(String expectedBatchHash) {
        return this.computeBatchHash().equals(expectedBatchHash);
    }
}

4.2 数据去重策略

数据重复是个老问题。网络重传、消费者重启、生产者重试，都会导致同一条数据被发送多次。去重策略的选择，取决于你对「精确一次」的要求有多高。

策略	原理	适用场景	代价
业务主键去重	用订单ID、用户ID等业务字段判断重复	有明确业务唯一键的场景	需要存储历史主键，内存/磁盘开销
消息ID去重	生产者生成全局唯一ID，消费者按ID去重	分布式系统，无业务主键	需要ID生成服务，额外网络开销
窗口去重	在时间窗口内去重，窗口外不再检查	实时流处理，允许少量重复	窗口大小难确定，太短漏去重，太长浪费内存
幂等写入	目标存储支持幂等操作，重复写入不产生新数据	数据库、消息队列等支持幂等的系统	对存储系统有要求，不是所有系统都支持

我个人最推荐的是「业务主键 + 窗口去重」的组合。举个例子：

// 基于Flink的窗口去重示例
DataStream<OrderEvent> stream = ...;

stream
    .keyBy(event -> event.getOrderId())  // 按业务主键分组
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .process(new DeduplicateFunction<>())
    .print();

// 在窗口内只保留第一条到达的记录
public class DeduplicateFunction extends ProcessWindowFunction<...> {
    @Override
    public void process(...) {
        // 只输出窗口内的第一条记录
        out.collect(elements.iterator().next());
    }
}

4.3 采集延迟控制

延迟控制，说白了就是「数据从产生到进入系统，花了多长时间」。延迟太高，实时分析就变成了离线分析。

我习惯把延迟分成三类：

• 端到端延迟：从数据产生到消费者看到数据的总时间
• 传输延迟：数据在网络中传输的时间
• 处理延迟：数据在采集端排队、校验、转换的时间

控制延迟的核心手段有三个：

1. 设置超时阈值：每条数据从采集到入队，超过N秒就报警
2. 背压机制：当下游处理不过来时，上游主动降速，防止数据积压
3. 优先级队列：对时效性要求高的数据（如交易数据）走高优通道

下面是一个简单的延迟监控代码片段：

// 在采集端记录时间戳
public class TimestampedRecord {
    private String data;
    private long produceTimestamp;  // 数据产生时间
    private long collectTimestamp;  // 采集端接收时间
    
    public long getTransportDelay() {
        return collectTimestamp - produceTimestamp;
    }
}

// 消费者端计算端到端延迟
long endToEndDelay = System.currentTimeMillis() - record.getProduceTimestamp();
if (endToEndDelay > THRESHOLD_MS) {
    alertService.sendAlert("延迟超标: " + endToEndDelay + "ms");
}

4.4 异常数据标记与重试机制

异常数据是躲不掉的。格式错误、字段缺失、业务逻辑冲突……你想想看，每天几亿条数据，总有一些「怪胎」。关键不是避免异常，而是优雅地处理异常。

我总结了一套「三阶段处理法」：

阶段	动作	说明
第一阶段	标记异常	给异常数据打标签，记录异常原因、原始数据、时间戳
第二阶段	隔离存储	将异常数据写入「死信队列」或「异常表」，不阻塞主流程
第三阶段	重试/修复	根据异常类型决定重试策略：自动重试、人工介入、或丢弃

重试策略的代码示例：

public class RetryHandler {
    private static final int MAX_RETRIES = 3;
    private static final long BASE_DELAY_MS = 1000;
    
    public boolean processWithRetry(Record record) {
        for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
            try {
                return doProcess(record);
            } catch (Exception e) {
                if (attempt == MAX_RETRIES) {
                    // 最后一次失败，写入死信队列
                    deadLetterQueue.send(record, e);
                    return false;
                }
                // 指数退避：1s, 2s, 4s
                long delay = BASE_DELAY_MS * (long) Math.pow(2, attempt - 1);
                Thread.sleep(delay);
            }
        }
        return false;
    }
}

知识体系总览

下面这张图把本章的核心逻辑串起来了。你可以看到，数据从源头出发，经过完整性校验、去重、延迟监控，最后进入异常处理流程——每个环节都在为「高质量数据」服务。