【架构实战】实时计算架构：Flink处理亿级数据流
## 一、离线报表晚一天，决策就晚一天 2020年，我们的数据报表是T+1： - 每天凌晨跑批 - 早上8点出报表 - 运营看昨天的数据 **问题：** - 618大促时，销量激增 - 运营想看实时数据调整策略 - 但只能看昨天的数据 - 错过最佳调整时机 引入实时计算后： - 数据秒级更新 - 实时大屏展示 - 实时预警 - 运营决策效率提升10倍 --- ## 二、实时计算架构 ### 2.1 Lambda架构 ``` ┌─────────────────────────────────────────────────────────────────┐ │ Lambda架构 │ │ │ │ 数据源 → Kafka ─┬→ Flink实时处理 → 实时结果 │ │ │ │ │ └→ HDFS离线存储 → Spark离线处理 → 离线结果 │ │ │ │ 实时结果 ─┬→ 合并 → 服务层 │ │ 离线结果 ─┘ │ │ │ │ 优点：容错性好，离线数据修正实时数据 │ │ 缺点：维护两套系统，复杂度高 │ │ │ └──────────────────────────────────────────────────────────────────┘ ``` ### 2.2 Kappa架构 ``` ┌─────────────────────────────────────────────────────────────────┐ │ Kappa架构 │ │ │ │ 数据源 → Kafka → Flink实时处理 → 结果存储 → 服务层 │ │ │ │ 特点： │ │ - 只维护一套实时处理系统 │ │ - Kafka作为唯一数据源 │ │ - 通过重放历史数据实现重新计算 │ │ │ │ 优点：架构简单，维护成本低 │ │ 缺点：历史数据重放成本高 │ │ │ └──────────────────────────────────────────────────────────────────┘ ``` --- ## 三、Flink核心应用 ### 3.1 实时数据大屏 ```java /** * 实时订单统计 */ public class OrderStatisticsJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 1. Kafka数据源 Properties props = new Properties(); props.setProperty("bootstrap.servers", "kafka:9092"); props.setProperty("group.id", "order-statistics"); FlinkKafkaConsumer source = new FlinkKafkaConsumer<>( "order-topic", new OrderDeserializer(), props ); DataStream orders = env.addSource(source); // 2. 实时统计 orders // 按时间窗口分组（每5秒一个窗口） .keyBy(Order::getProductId) .window(TumblingProcessingTimeWindows.of( Time.seconds(5))) // 聚合计算 .aggregate(new OrderAggregateFunction()) // 写入Redis .addSink(new RedisSink<>(...)); env.execute("Order Statistics"); } } /** * 订单聚合函数 */ public class OrderAggregateFunction implements AggregateFunction { @Override public OrderAccumulator createAccumulator() { return new OrderAccumulator(); } @Override public OrderAccumulator add(Order order, OrderAccumulator acc) { acc.addAmount(order.getAmount()); acc.addCount(1); return acc; } @Override public OrderResult getResult(OrderAccumulator acc) { return new OrderResult( acc.getProductId(), acc.getCount(), acc.getAmount(), System.currentTimeMillis() ); } } ``` ### 3.2 实时风控 ```java /** * 实时风控监测 */ public class RiskControlJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 1. 消费用户行为流 DataStream behaviors = env .addSource(new KafkaSource<>("user-behavior")); // 2. 实时规则检测 behaviors .keyBy(UserBehavior::getUserId) .process(new RiskDetectionProcessFunction()) .addSink(new AlertSink()); env.execute("Risk Control"); } } /** * 风险检测函数 */ public class RiskDetectionProcessFunction extends KeyedProcessFunction { // 状态：用户最近行为列表 private ListState behaviorState; @Override public void open(Configuration parameters) { ListStateDescriptor descriptor = new ListStateDescriptor<>( "behaviors", UserBehavior.class); behaviorState = getRuntimeContext().getListState(descriptor); } @Override public void processElement(UserBehavior behavior, Context ctx, Collector out) throws Exception { // 1. 添加到状态 behaviorState.add(behavior); // 2. 获取最近1分钟的行为 List recentBehaviors = new ArrayList<>(); for (UserBehavior b : behaviorState.get()) { if (System.currentTimeMillis() - b.getTimestamp() b.getType() == BehaviorType.LOGIN_FAIL) .count(); if (loginFailCount > 5) { out.collect(new RiskAlert( behavior.getUserId(), "LOGIN_FAIL_TOO_MANY", loginFailCount)); } // 规则2：1分钟内下单超过10笔 long orderCount = recentBehaviors.stream() .filter(b -> b.getType() == BehaviorType.ORDER) .count(); if (orderCount > 10) { out.collect(new RiskAlert( behavior.getUserId(), "ORDER_TOO_FREQUENT", orderCount)); } // 4. 注册定时器清理过期状态 ctx.timerService().registerProcessingTimeTimer( System.currentTimeMillis() + 60000); } } ``` ### 3.3 实时ETL ```java /** * 实时数据同步 */ public class RealtimeETLJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 1. Canal数据变更流 DataStream canalStream = env .addSource(new CanalSource("order-table")); // 2. 数据清洗转换 DataStream orders = canalStream .filter(entry -> entry.getEventType() == INSERT || entry.getEventType() == UPDATE) .map(entry -> convertToOrder(entry)); // 3. 写入多个目标 // 3.1 写入ES（搜索） orders.addSink(new ElasticsearchSink<>(...)); // 3.2 写入Redis（缓存） orders.addSink(new RedisSink<>(...)); // 3.3 写入ClickHouse（分析） orders.addSink(new ClickHouseSink<>(...)); env.execute("Realtime ETL"); } } ``` --- ## 四、状态管理 ### 4.1 状态后端配置 ```java // 配置RocksDB状态后端 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 1. 使用RocksDB（大状态） env.setStateBackend(new EmbeddedRocksDBStateBackend()); // 2. 配置Checkpoint env.enableCheckpointing(60000); // 每60秒checkpoint一次 env.getCheckpointConfig().setCheckpointingMode( CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); env.getCheckpointConfig().setCheckpointTimeout(60000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); env.getCheckpointConfig().enableExternalizedCheckpoints( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 3. 配置Savepoint路径 env.getCheckpointConfig().setCheckpointStorage( "hdfs:///flink/checkpoints"); ``` ### 4.2 状态恢复 ```java // 从Savepoint恢复 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStateBackend(new EmbeddedRocksDBStateBackend()); // 指定Savepoint路径 String savepointPath = "hdfs:///flink/savepoints/savepoint-123"; env.getCheckpointConfig().setCheckpointStorage(savepointPath); // 从Savepoint恢复执行 env.execute("My Job"); ``` --- ## 五、踩坑实录 ### 坑1：背压问题 > **问题**：下游消费慢，导致上游数据积压。 **解决方案**： ```java // 1. 增加并行度 env.setParallelism(8); // 2. 异步Sink orders.addSink(new AsyncRedisSink<>()); // 3. 调整缓冲区大小 env.getConfig().setBufferTimeout(100); ``` ### 坑2：数据倾斜 > **问题**：某些key数据量大，导致热点。 **解决方案**： ```java // 1. 加盐打散 orders.keyBy(order -> order.getUserId() + "_" + RandomUtils.nextInt(0, 10)) .window(...) // 2. 两阶段聚合 orders.keyBy(order -> order.getUserId() + "_" + salt) .window(...) .aggregate(...) // 第一阶段：局部聚合 .keyBy(...) .window(...) .aggregate(...); // 第二阶段：全局聚合 ``` ### 坑3：状态过大 > **问题**：状态无限增长，内存溢出。 **解决方案**： ```java // 1. 使用TTL清理过期状态 StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.hours(24)) .setUpdateType(UpdateType.OnCreateAndWrite) .setStateVisibility(StateVisibility.NeverReturnExpired) .build(); ValueStateDescriptor descriptor = new ValueStateDescriptor<>("state", String.class); descriptor.enableTimeToLive(ttlConfig); // 2. 使用RocksDB状态后端 env.setStateBackend(new EmbeddedRocksDBStateBackend()); ``` ### 坑4：Exactly-Once不生效 > **问题**：重启后数据重复或丢失。 **解决方案**： ```java // 1. 确保Sink支持事务 FlinkKafkaProducer sink = new FlinkKafkaProducer<>( "topic", new SimpleStringSchema(), props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE // 开启事务 ); // 2. 配置Checkpoint env.enableCheckpointing(60000); env.getCheckpointConfig().setCheckpointingMode( CheckpointingMode.EXACTLY_ONCE); ``` --- ## 六、最佳实践 ### 6.1 Flink调优清单 ``` Flink调优清单： □ 并行度 □ 与Kafka分区数匹配 □ 根据数据量调整 □ 内存 □ TaskManager内存充足 □ Network缓冲区足够 □ 状态 □ 大状态用RocksDB □ 配置状态TTL □ Checkpoint □ 间隔合理（1-5分钟） □ 超时时间充足 □ 保留Checkpoint □ 背压 □ 监控背压指标 □ 及时扩容 ``` --- ## 七、总结 实时计算核心要点： | 组件 | 作用 | 选型 | |------|------|------| | 消息队列 | 数据缓冲 | Kafka | | 计算引擎 | 实时处理 | Flink | | 状态存储 | 状态管理 | RocksDB | | 结果存储 | 结果输出 | Redis/ES | **血的教训：** > 实时计算是数据的"高速公路"，搭建好了，数据才能流得快。 --- *个人观点，仅供参考*
