ETL 架构演进：从批处理到实时流处理实战

数据清洗策略

清洗是保证数据质量的关键步骤，通常包括去重、空值处理和异常值过滤。

from pyspark.sql.functions import regexp_replace, to_timestamp

def clean_data(df):
    # 去除重复行
    df = df.dropDuplicates()
    
    # 处理空值，填充默认值
    df = df.fillna({"phone": "UNKNOWN", "address": "UNKNOWN"})
    
    # 数据标准化，例如手机号只保留数字
    df = df.withColumn("phone", regexp_replace(col("phone"), "[^0-9]", ""))
    df = df.withColumn("order_time", to_timestamp(col("order_time"), "yyyy-MM-dd HH:mm:ss"))
    
    # 异常值处理，过滤不合理金额
    df = df.filter((col("amount") > 0) & (col("amount") < 1000000))
    return df

实时流处理 ETL

当业务对时效性要求变高时，流处理成为必然选择。Flink 凭借其低延迟和高吞吐特性，是目前实时计算的首选。

Flink 实时优势

• 低延迟：秒级甚至毫秒级响应
• 资源均匀：持续消费，避免批处理的波峰波谷
• 状态管理：内置 Checkpoint 机制，保障数据一致性

WordCount 基础示例

理解流处理最好从经典的 WordCount 开始，这展示了基本的 Source、Transformation 和 Sink 流程。

public class FlinkWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // 读取数据源
        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // 转换和聚合
        DataStream<WordCount> counts = text
            .flatMap((line, out) -> {
                for (String word : line.split("\\s")) {
                    out.collect(new WordCount(word, 1));
                }
            })
            .keyBy("word")
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
            .sum("count");

        // 输出
        counts.print();
        env.execute("WordCount");
    }
}

实时数据同步

在生产环境中，我们常需要将 Kafka 中的数据直接同步到 Hive 或数仓中。

public class KafkaToHiveETL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000); // 每分钟检查点
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);

        KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers("kafka:9092")
            .setGroupId("order-etl-consumer")
            .setTopics("orders")
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .setStartingOffsets(OffsetsInitializer.committedOffsets())
            .build();

        DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        // 解析 JSON 并过滤
        DataStream<Order> orders = stream
            .map(json -> JSON.parseObject(json, Order.class))
            .filter(order -> order.getAmount() > 0);

        // 写入 Hive
        FlinkHiveConnector.writeToHive(orders, "warehouse.orders");
        env.execute("KafkaToHiveETL");
    }
}

实时聚合

除了简单的同步，实时聚合更能体现流计算的价值，比如统计店铺近 5 分钟的订单情况。

public class RealTimeShopStats {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 读取 Kafka
        DataStream<Order> orders = env.addSource(new FlinkKafkaConsumer<>("orders", new OrderDeserializationSchema(), KafkaConfig.getProperties()));

        // 实时统计：滑动窗口
        DataStream<ShopStats> stats = orders
            .keyBy(Order::getShopId)
            .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
            .aggregate(new ShopStatsAggregator());

        // 输出到 Kafka（供下游消费）
        stats.addSink(new FlinkKafkaProducer<>("shop-realtime-stats", new ShopStatsSerializationSchema(), KafkaConfig.getProperties()));
        
        // 输出到 Redis（供仪表盘查询）
        stats.addSink(new RedisSink<>(redisConfig));
        
        env.execute("RealTimeShopStats");
    }
}

// 聚合器实现
public class ShopStatsAggregator implements AggregateFunction<Order, ShopStatsAccumulator, ShopStats> {
    @Override
    public ShopStatsAccumulator createAccumulator() {
        return new ShopStatsAccumulator();
    }

    @Override
    public ShopStatsAccumulator add(Order value, ShopStatsAccumulator acc) {
        acc.orderCount++;
        acc.totalAmount += value.getAmount();
        acc.maxAmount = Math.max(acc.maxAmount, value.getAmount());
        return acc;
    }

    @Override
    public ShopStats getResult(ShopStatsAccumulator acc) {
        return new ShopStats(acc.getShopId(), acc.orderCount, acc.totalAmount, 
            acc.totalAmount / acc.orderCount, acc.maxAmount);
    }

    @Override
    public ShopStatsAccumulator merge(ShopStatsAccumulator a, ShopStatsAccumulator b) {
        a.orderCount += b.orderCount;
        a.totalAmount += b.totalAmount;
        a.maxAmount = Math.max(a.maxAmount, b.maxAmount);
        return a;
    }
}

Lambda 架构设计

Lambda 架构试图兼顾批处理的准确性和流处理的低延迟，通过'批处理层 + 速度层'来覆盖所有场景。

架构分层

• 批处理层：维护全量历史数据，计算准确但延迟高
• 速度层：处理增量数据，计算快但可能不精确
• 服务层：合并两层结果，对外提供统一查询接口

实现示例

在服务层查询时，我们需要将昨天的全量数据和今天的实时数据进行合并。

-- 批处理层 - 计算全量指标
SELECT shop_id, COUNT(*) as order_count, SUM(amount) as total_amount, MAX(amount) as max_amount 
FROM orders 
WHERE order_date >= '2023-01-01' 
GROUP BY shop_id, order_date;

-- 服务层 - 合并查询
SELECT 
    a.shop_id, 
    a.total_amount + COALESCE(b.realtime_amount, 0) as total_amount, 
    a.order_count + COALESCE(b.realtime_count, 0) as order_count 
FROM (
    SELECT shop_id, total_amount, order_count 
    FROM warehouse.shop_daily_stats_batch 
    WHERE shop_id = {shop_id} AND order_date = CURRENT_DATE - 1
) a 
LEFT JOIN (
    SELECT shop_id, SUM(total_amount) as realtime_amount, SUM(order_count) as realtime_count 
    FROM warehouse.shop_realtime_stats 
    WHERE shop_id = {shop_id} 
    GROUP BY shop_id
) b ON a.shop_id = b.shop_id;

Kappa 架构演进

Kappa 架构是对 Lambda 的简化，主张'一切皆流'，抛弃了专门的批处理层，完全依赖流处理引擎。

核心思想

• 只使用流处理管道
• 历史数据通过重放（Replay）Kafka 消息来实现
• 架构更简单，维护成本更低

全量历史重计算

如果需要重新计算历史数据，只需修改 Kafka 的消费起始位点即可。

public class KappaETL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1. 从 Kafka 读取历史数据（全量重放）
        DataStream<Order> historicalOrders = env
            .addSource(new FlinkKafkaConsumer<>("orders", new OrderDeserializationSchema(), KafkaConfig.getProperties()))
            .setStartFromEarliest();

        // 2. 从 Kafka 读取实时数据
        DataStream<Order> realtimeOrders = env.addSource(new FlinkKafkaConsumer<>("orders", new OrderDeserializationSchema(), KafkaConfig.getProperties()));

        // 3. 合并数据源
        DataStream<Order> allOrders = historicalOrders.union(realtimeOrders);

        // 4. 流处理计算
        DataStream<ShopStats> stats = allOrders
            .keyBy(Order::getShopId)
            .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
            .aggregate(new ShopStatsAggregator());

        // 5. 输出结果
        stats.addSink(new RedisSink<>(redisConfig));
        env.execute("KappaETL");
    }
}

工具选型对比

总结与建议

ETL 架构的演进本质上是数据时效性与系统复杂度之间的权衡。

• 批处理：适合对延迟不敏感的历史分析场景
• 实时处理：适合需要秒级反馈的监控和风控场景
• Lambda：适合既有强一致性要求又有实时需求的复杂系统
• Kappa：适合希望简化运维、统一技术栈的团队

选型建议：

1. 离线分析：优先使用 Spark 批处理，生态成熟稳定
2. 实时监控：Flink 流处理是目前的最佳实践
3. 复杂场景：根据团队技术储备决定采用 Lambda 还是 Kappa，注意不要过度设计