3、数据过滤与空值处理：把脏数据挡在门外

数据清洗里最基础也最频繁的操作是什么？我个人的答案是过滤与空值处理。你想想看，不管数据从哪来——Kafka、文件、数据库——总会有一些「垃圾」混进来。字段缺失、格式错乱、值为 null，这些如果不处理，下游的计算结果就会出问题。

这一节，我们就来聊聊怎么用 Flink 的 filter 算子把无效数据筛掉，怎么处理烦人的 null 值，以及一些我踩过的坑。

3.1 filter 算子：最简单的数据守门员

filter 算子的用法很直观。它接收一个布尔表达式，返回 true 的数据保留，false 的丢弃。说白了，就是给数据流装了个筛子。

// 过滤掉年龄小于 0 或大于 150 的非法数据
DataStream<User> validUsers = userStream
    .filter(user -> user.getAge() >= 0 && user.getAge() <= 150);

我在项目中遇到过一种情况：上游系统偶尔会发送一些「幽灵数据」——所有字段都是空字符串，但又不是 null。这种数据用常规的 null 检查根本抓不住。后来我加了个组合条件：

DataStream<Event> cleanStream = rawStream
    .filter(event -> {
        // 排除全空字段的脏数据
        if (event.getUserId() == null || event.getUserId().trim().isEmpty()) {
            return false;
        }
        if (event.getEventType() == null || event.getEventType().trim().isEmpty()) {
            return false;
        }
        return true;
    });

3.2 null 值的三种处理策略

null 值是脏数据里的「常客」。我个人习惯把 null 处理分成三个层次：

策略一：直接过滤

如果某个字段是业务主键，比如订单 ID、用户 ID，那它不能为 null。遇到这种情况，直接过滤掉是最安全的。

// 订单 ID 为 null 的数据毫无价值
DataStream<Order> validOrders = orderStream
    .filter(order -> order.getOrderId() != null);

策略二：三元表达式兜底

对于非关键字段，比如用户昵称、备注信息，用三元表达式给个默认值就够了。这样做的好处是代码简洁，不会因为 null 引发空指针。

DataStream<User> enrichedUsers = userStream
    .map(user -> {
        String name = user.getName() != null ? user.getName() : "匿名用户";
        Integer age = user.getAge() != null ? user.getAge() : 0;
        return new User(user.getUserId(), name, age);
    });

3.3 自定义空值处理函数

有些场景下，简单的兜底或过滤解决不了问题。举个例子，我曾经遇到一个日志系统，它的 timestamp 字段有时是 null，但可以从日志内容里解析出来。这时候就需要写一个自定义函数来处理。

public class NullValueProcessor implements MapFunction<LogEvent, LogEvent> {
    
    @Override
    public LogEvent map(LogEvent event) throws Exception {
        // 修复 null 的时间戳
        if (event.getTimestamp() == null) {
            String extractedTime = extractTimestampFromMessage(event.getMessage());
            if (extractedTime != null) {
                event.setTimestamp(Long.parseLong(extractedTime));
            } else {
                // 实在修复不了，用当前时间兜底
                event.setTimestamp(System.currentTimeMillis());
            }
        }
        
        // 修复 null 的 IP 地址
        if (event.getIpAddress() == null || event.getIpAddress().isEmpty()) {
            event.setIpAddress("0.0.0.0");
        }
        
        return event;
    }
    
    private String extractTimestampFromMessage(String message) {
        // 假设日志格式: [2024-01-15 10:30:00] ERROR ...
        if (message != null && message.startsWith("[")) {
            int endIndex = message.indexOf("]");
            if (endIndex > 0) {
                String timeStr = message.substring(1, endIndex);
                // 这里可以调用日期解析工具类
                return parseTimeToTimestamp(timeStr);
            }
        }
        return null;
    }
}

// 使用
DataStream<LogEvent> fixedStream = logStream
    .map(new NullValueProcessor());

这种自定义函数的优势在于：你可以把修复逻辑封装起来，方便复用和测试。而且当业务规则变化时，只需要改一个地方。

3.4 实战中的避坑指南

做空值处理这几年，我踩过不少坑。挑几个典型的说说：

• 别用 0 或空字符串代替 null —— 我曾经用 0 代替 null 的年龄字段，结果下游计算平均年龄时，把一堆 0 算进去了，导致均值严重偏低。后来我改用 -1 作为缺失标识，并在计算时显式排除。
• 注意 filter 后的数据量变化 —— 如果过滤掉的数据太多，可能会影响下游窗口的触发。比如事件时间窗口，数据量骤减可能导致 watermark 长时间不推进。
• null 检查要放在最前面 —— 在 filter 或 map 里，先检查 null，再做其他业务逻辑。否则一旦出现空指针，整个任务可能挂掉。

3.5 知识体系总览

下面这张图把本章的核心逻辑串起来了。你可以看到，数据从源头进来，经过 filter 筛掉明显无效的数据，再通过空值处理策略（过滤、兜底、自定义函数）进行精细化清洗，最终得到干净的数据流。

嗯，到这里，数据过滤和空值处理的核心内容就差不多了。记住一个原则：宁可多写几行检查代码，也不要让脏数据溜进下游。我在生产环境里见过太多因为 null 处理不当导致的线上事故，有些甚至要回滚数据才能修复。