📚 课程目录第3章 Flink批流一体架构

3. Flink 批流一体架构：Batch on Streaming 原理与 API 融合

在支付风控漂移检测场景中，数据源天然兼具“实时流”与“历史批”双重属性：交易流水是持续到达的流，而离线特征库、黑名单快照、模型基线则是静态批次。Flink 1.12+ 的批流一体架构彻底终结了“两套引擎、两套代码”的割裂时代，让风控工程师能用同一套 DataStream API 处理实时与历史数据，大幅降低漂移检测的开发和运维成本。

3.1 Batch on Streaming 核心原理

Flink 的批处理并非独立运行时模式，而是将批作业视为“有界流”（Bounded Stream）在流引擎上执行。这一设计被称为 Batch on Streaming，其本质是复用流式执行框架的全部能力（状态管理、容错、精确一次语义），仅通过输入数据的有界性触发优化执行策略。

对比维度	传统批处理（MapReduce/Spark Batch）	Flink Batch on Streaming
执行引擎	独立批处理引擎	统一流引擎，批作业视为有界流
调度策略	阶段式（Stage）调度，依赖 Shuffle	流水线式（Pipelined）调度，支持链式执行
容错机制	任务失败重算整个 Stage	基于 Checkpoint 的增量容错，精确一次
数据倾斜处理	需手动调优分区	动态负载均衡 + 自适应 Hash Join
资源利用率	批次间资源释放/申请开销大	TaskManager 常驻，流批作业共享 Slot

  风控场景收益： 漂移检测中的“T+1 离线特征比对”与“实时交易评分”可在同一 Flink 集群运行，共享状态后端与内存管理，避免数据跨集群拷贝。

3.2 DataSet API 与 DataStream API 的融合演进

Flink 1.12 是批流 API 融合的里程碑版本。DataSet API 被标记为“软弃用”（Soft Deprecated），官方推荐所有新作业统一使用 DataStream API。融合后的 API 层次如下：

统一 Source/Sink 接口： FileSource、KafkaSource 等 Connector 同时支持有界（批）和无界（流）模式，通过 Boundedness 属性区分。
统一算子语义： map、filter、keyBy、window 在批模式下自动退化为全量计算，无需切换 API。
Table/SQL 全面批流一体： 同一张 DDL 表定义，通过 scan（批）或 stream（流）模式执行查询。
DataSet 遗留算子迁移： GroupReduce、GroupCombine 等批专用算子被 KeyedProcessFunction + 定时器替代。

// Flink 1.12+ 批流统一示例：同一份代码处理实时流和历史文件
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 实时流：Kafka 无界数据
DataStream<Transaction> realtimeStream = env
    .fromSource(KafkaSource.<Transaction>builder()
        .setBoundedness(Boundedness.CONTINUOUS_UNBOUNDED)
        .build(), "kafka-source");

// 批数据：HDFS 有界文件
DataStream<Transaction> batchStream = env
    .fromSource(FileSource.forRecordStreamFormat(...)
        .build(), "file-source");

// 统一处理逻辑：漂移检测特征提取
DataStream<DriftFeature> features = realtimeStream
    .union(batchStream)
    .keyBy(tx -> tx.getMerchantId())
    .process(new DriftFeatureExtractor());

env.execute("Unified Drift Detection Job");

3.3 Flink 1.12+ 批流统一架构图

下图展示了 Flink 批流一体的运行时架构：批作业与流作业共享同一个执行引擎，仅在数据源端通过 Boundedness 标记区分有界/无界，下游算子透明执行。

3.4 批流一体在漂移检测中的实践模式

支付风控漂移检测通常需要同时处理“实时交易流”和“历史特征批次”。批流一体架构提供了三种典型组合模式：

模式	描述	适用场景	Flink 实现方式
流批混合 Join	实时流与静态批次做 Interval Join 或 Lookup Join	实时交易 <-> 黑名单快照	`DataStream` + `RichAsyncFunction` 加载批数据
批计算预热状态	先用批作业初始化状态，再切换为流处理	模型基线加载、历史特征预计算	`BoundedStream` → `saveState()` → 无界流恢复
统一特征管道	同一 DAG 内同时消费有界/无界数据源，输出统一特征	漂移检测特征提取	`union()` 或 `connect()` 合并流批

最佳实践： 在漂移检测作业中，优先使用 DataStream API + FileSource 读取历史特征文件，避免引入 DataSet API 造成 API 混用和维护成本。Flink 1.14+ 已完全移除 DataSet API 的运行时依赖。

3.5 批流一体下的状态与容错差异

虽然批作业复用流引擎，但在状态管理和容错策略上存在关键差异，风控作业需针对性配置：

状态后端选择： 批作业默认使用 HashMapStateBackend（堆内存），流作业推荐 RocksDBStateBackend。批流混合作业需根据算子实际状态大小动态调整。
Checkpoint 行为： 批作业默认不启用 Checkpoint（因为数据有界，失败可重算），但若批作业参与流批混合 DAG，需手动启用 Checkpoint 以保证一致性。
Watermark 处理： 批数据源不产生 Watermark，与流数据 union 时需通过 assignTimestampsAndWatermarks 为批数据注入固定 Watermark（如 WatermarkStrategy.forMonotonousTimestamps()）。

// 批流混合作业的 Checkpoint 与 Watermark 配置示例
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 启用 Checkpoint（即使包含批数据源）
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink-checkpoints");

// 批数据源注入单调递增 Watermark
DataStream<Transaction> batchStream = env
    .fromSource(fileSource, 
        WatermarkStrategy.<Transaction>forMonotonousTimestamps()
            .withTimestampAssigner((event, ts) -> event.getEventTime()),
        "batch-source");

// 实时流使用事件时间 Watermark
DataStream<Transaction> realtimeStream = env
    .fromSource(kafkaSource,
        WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(5)),
        "realtime-source");

注意： 在支付风控场景中，批流混合作业的 Checkpoint 间隔不宜过短（建议 ≥ 30 秒），避免批数据源的全量扫描触发频繁快照，导致 HDFS 写入压力过大。

3.6 从 DataSet 到 DataStream 的迁移路径

对于仍在使用 DataSet API 的遗留风控作业，Flink 1.12+ 提供了清晰的迁移路线：

替换 Source： 将 ExecutionEnvironment.readTextFile() 替换为 StreamExecutionEnvironment.fromSource(FileSource)。
替换算子： GroupReduceFunction → KeyedProcessFunction + 状态 ListState；GroupCombine → window().aggregate()。
替换 Sink： DataSet.output() → DataStream.sinkTo()，使用统一的 StreamingFileSink 或 JdbcSink。
验证语义： 批作业迁移后需验证结果一致性，特别是涉及 sortPartition、first 等批特有算子时，需用 keyBy + windowAll 模拟。

// 迁移示例：DataSet 的 GroupReduce → DataStream 的 KeyedProcessFunction
// 旧 DataSet API
DataSet<Tuple2<String, Double>> result = env
    .readTextFile("hdfs:///history")
    .groupBy(0)
    .reduceGroup(new OldGroupReduce());

// 新 DataStream API
DataStream<Tuple2<String, Double>> result = streamEnv
    .fromSource(fileSource, "history")
    .keyBy(t -> t.f0)
    .process(new KeyedProcessFunction<String, Transaction, Tuple2<String, Double>>() {
        private ValueState<Double> sumState;
        @Override
        public void open(Configuration parameters) {
            sumState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("sum", Types.DOUBLE));
        }
        @Override
        public void processElement(Transaction value, Context ctx, Collector<Tuple2<String, Double>> out) {
            Double current = sumState.value() == null ? 0.0 : sumState.value();
            sumState.update(current + value.getAmount());
        }
    });

通过上述迁移，风控团队可以彻底告别两套 API 的维护负担，在统一的 DataStream API 上构建端到端的漂移检测管道，实现“开发一次，流批皆可运行”的最终目标。