Flink实时特征工程

一、写在前面：为什么实时特征工程需要Flink？

1.1 传统批处理的痛点

• 场景错配：用户行为数据（点击/加购）天然是流式的，T+1统计导致特征时效性差（昨天用户加购的商品，今天才进入推荐系统）
• 资源浪费：凌晨集中跑批时集群压力大，白天资源闲置
• 业务响应慢：新活动效果评估要等次日，无法实时调整策略

1.2 Flink的杀手锏

• 真正的流处理：数据像水流一样持续处理（对比Spark Streaming的微批）
• 事件时间语义：处理乱序数据不乱（双十一订单可能延迟到达）
• 状态管理：记住用户最近30次点击（没有Redis也能玩转上下文特征）
• Exactly-Once：金融级精度保障（不会重复扣款）
• 高吞吐低延迟：单节点每秒处理百万条，亚秒级延迟（实测每秒处理20W条用户行为日志，P99延迟<500ms）

---

二、Flink实时特征工程架构解剖

2.1 典型架构图

[Kafka] --> [Flink SQL ETL] --> [特征存储(HBase/Redis)] --> [在线服务]
         |--> [实时数仓(Doris)] 
         |--> [监控告警(Prometheus)]

2.2 核心组件详解

• Source层：

  • Kafka（主流选择，需关注Consumer Group偏移量管理）
  • 自定义Source（兼容埋点SDK直传场景）
  • CDC（MySQL Binlog实时捕获，适合订单特征）

• Processing层：

  • Operator Chains：算子链优化（减少序列化开销）
  • Watermark机制：对付延迟数据的"时间沙漏"（允许乱序3秒）
  • State Backend：
    • MemoryStateBackend（本地调试用）
    • RocksDB（生产标配，支持增量Checkpoint）
    • 自定义状态后端（对接外部存储如Cassandra）

• Sink层：

  • Keyed存储：用户维度特征入Redis（String/Hash结构）
  • OLAP存储：Doris/ClickHouse存聚合特征（供BI实时分析）
  • 回撤流处理：维表JOIN时处理更新的技巧（避免特征漂移）

---

三、时间窗口的十八般武艺

3.1 滚动窗口（Tumbling）

• 适用场景：每分钟UV统计
• 代码示例：

windowedStream = dataStream
  .keyBy(UserBehavior::getUserId)
  .window(TumblingEventTimeWindows.of(Time.minutes(1)));

3.2 滑动窗口（Sliding）

• 特殊配置：窗口大小=10min，滑动步长=1min（内存消耗需警惕）
• 优化技巧：复用窗口计算结果（增量聚合函数）

3.3 会话窗口（Session）

• 动态特性：用户两次操作间隔超5min则拆分会话
• 参数调优：gap时间设置影响内存占用（长会话需状态清理策略）

3.4 全局窗口（Global）

• 高阶用法：配合触发器实现TopN统计（每100条触发计算）

---

四、状态管理的生存指南

4.1 状态类型

• ValueState：存储单个值（用户最后登录时间）
• ListState：存储列表（最近10次搜索关键词）
• MapState：键值对（商品ID -> 点击次数）
• AggregatingState：自定义聚合（维护复杂统计对象）

4.2 状态容错

• Checkpoint流程：

  1. JobManager发起Checkpoint请求
  2. Source插入Barrier（类似水流中的分隔标记）
  3. 算子异步快照状态
  4. 全链路确认后提交

• 恢复策略：

  • 全量恢复：从最近完成的Checkpoint重启
  • 增量恢复（RocksDB专属优势）

4.3 状态过期

• TTL配置：

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .setUpdateType(OnCreateAndWrite))
    .cleanupInRocksdbCompactFilter(1000)
    .build();

---

五、实战：电商用户画像特征生成

5.1 需求拆解

• 短期兴趣：最近1小时点击类目分布
• 长期偏好：过去30天购买品牌TOP3
• 实时特征：当前购物车金额

5.2 代码实现

DataStream<UserBehavior> stream = env.addSource(kafkaSource);

// 实时计算购物车总金额
SingleOutputStreamOperator<CartTotal> cartFeatures = stream
    .filter(behavior -> "cart".equals(behavior.getAction()))
    .keyBy(UserBehavior::getUserId)
    .process(new CartAggregator());

// 近1小时类目偏好（滑动窗口）
SingleOutputStreamOperator<CategoryPref> cateFeatures = stream
    .filter(behavior -> "click".equals(behavior.getAction()))
    .keyBy(UserBehavior::getUserId)
    .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
    .aggregate(new CategoryAggregate());

// 维表关联商品信息
cartFeatures.connect(itemDimStream)
    .keyBy(cart -> cart.getItemId(), dim -> dim.getItemId())
    .process(new ItemEnrichment());

5.3 性能优化

• 旁路缓存：维表查询增加Guava Cache（减少DB压力）
• 异步IO：提升Redis查询吞吐（实测QPS提升8倍）
• 批量写入：攒批10ms或满100条写入HBase

---

六、生产环境避坑大全

6.1 反压定位

• 监控指标：outQueueUsage > 0.8 告警
• 排查步骤：
  1. Checkpoint时长是否激增
  2. 查看火焰图定位热点代码
  3. 检查外部系统吞吐（如HBase是否限流）

6.2 内存调优

• 配置参数：

taskmanager.memory.process.size: 4096m
taskmanager.memory.managed.fraction: 0.4

• GC优化：G1垃圾回收器 + 适当增大堆内存

6.3 数据倾斜

• 预聚合：在LocalKeyBy阶段做局部聚合
• 加盐打散：对热点Key添加随机后缀
• 维度升级：将用户维特征提升到设备维度

---

七、Flink 1.16 新特性应用

7.1 自适应批处理

• 动态调整：根据数据量自动切换流/批模式
• 资源节省：小批量数据减少Checkpoint开销

7.2 混合源支持

• 统一API：同一Job中处理有界/无界数据

FileSource<String> source = FileSource
    .forRecordStreamFormat(new TextLineFormat(), Path.fromLocalFile(testDir))
    .build();

7.3 增强的窗口语义

• 会话窗口优化：支持基于计数的会话切割
• 窗口合并策略：自定义相邻窗口合并逻辑

---

八、实践

8.1 监控体系搭建

• 核心指标：
  • 消费延迟（Kafka Lag）
  • Checkpoint成功率
  • 算子背压比例
• 可视化方案：Grafana + Prometheus 自定义看板

8.2 混沌工程实践

• 故障注入：
  • 随机Kill TaskManager
  • 模拟网络分区
  • 制造高延迟数据

8.3 成本控制

• 自动扩缩容：根据Kafka堆积量动态调整并行度
• 冷热分离：7天前的特征转存至OSS降低成本

---

参考文档：https://developer.aliyun.com/article/1176024