用 Flink DataStream API 搭建流式 ETL从无状态到有状态、从单流到连接流

1. 为什么用 DataStream 做 ETL？

ETL 的本质：从多源抽取 → 转换/富化 → 入库/下发。
在 Flink 里你有两条主路：

• Table/SQL API：声明式、上手快、适合标准 ETL 场景；
• DataStream API：灵活可编程、对自定义序列化/复杂业务逻辑/低延迟控制更友好。

建议：以 SQL 为主，遇到复杂逻辑/时序/状态机型问题切到 DataStream。理解 DataStream 的底层模型，会让你在 SQL 瓶颈时有“降级控制”的抓手。

2. 可流式的数据与序列化选型

• Flink 原生序列化支持：基础类型（String/Long/…）、Tuples（Tuple0~Tuple25）、POJOs（满足 public 类、无参构造、字段可见或有 getter/setter）。
• 其他类型回退 Kryo；也可使用 Avro（强 schema、跨语言好）。
• 建议：业务事件→POJO/Avro；原型/临时脚本→Tuple 均可。

Tuple 示例：

Tuple2<String, Integer> person = Tuple2.of("Fred", 35);
String name = person.f0; // 下标从0开始
Integer age  = person.f1;

POJO 示例（支持 schema 演进）：

public class Person {public String name;public Integer age;public Person() {}public Person(String name, Integer age) { this.name = name; this.age = age; }
}

3. 无状态转换：map / flatMap 富化与拆分

场景：把出租车行程（TaxiRide）按经纬度映射到 ~100×100m 网格，补充 startCell/endCell 字段。

public static class EnrichedRide extends TaxiRide {public int startCell, endCell;public EnrichedRide() {}public EnrichedRide(TaxiRide r) {this.rideId = r.rideId; this.isStart = r.isStart; /*...*/this.startCell = GeoUtils.mapToGridCell(r.startLon, r.startLat);this.endCell   = GeoUtils.mapToGridCell(r.endLon,   r.endLat);}
}DataStream<TaxiRide> rides = env.addSource(new TaxiRideSource(...));// 1→1 用 map
DataStream<EnrichedRide> enriched = rides.filter(new RideCleansingSolution.NYCFilter()).map((MapFunction<TaxiRide, EnrichedRide>) EnrichedRide::new);// 1→N 或 0→1 用 flatMap（可发零个或多个）
DataStream<EnrichedRide> enriched2 = rides.flatMap((ride, out) -> {if (new RideCleansing.NYCFilter().filter(ride)) out.collect(new EnrichedRide(ride));
});

要点：

• map()：严格一进一出。
• flatMap()：更自由，配合 Collector 发 0…N 个记录，更适合“过滤 + 拆分”。

4. Keyed Streams：按键分区与聚合

把相同 key 的事件路由到同一并行子任务，才能维护独立状态或做分组聚合。

enriched.keyBy(r -> r.startCell);

⚠️ 代价：每次 keyBy 都会触发 网络 shuffle（序列化+网络+反序列化）。
建议：合并相邻的算子链、减少不必要的重分区。

按键聚合示例：求每个起点单元的“最长行程（分钟）”

DataStream<Tuple2<Integer, Minutes>> minutesByStartCell = enriched.flatMap((ride, out) -> {if (!ride.isStart) {Minutes dur = new Interval(ride.startTime, ride.endTime).toDuration().toStandardMinutes();out.collect(Tuple2.of(ride.startCell, dur));}
});minutesByStartCell.keyBy(v -> v.f0) // startCell.maxBy(1)         // 对 duration 取最大.print();

隐式状态：maxBy 背后，Flink 为每个 key维护“当前最大值”。
风险：无界 key 数 → 状态无界膨胀。
建议：在窗口内聚合（滚动/滑动/会话窗口），或引入TTL/清理策略。

5. 有状态转换：让 Flink 管理状态（ValueState）

为什么交给 Flink：

• 本地访问（内存级速度）、自动 Checkpoint（容错）、RocksDB 托底（磁盘扩容）、横向扩展（重分布）。

去重示例：只保留每个 key 的第一条

public static class Deduplicator extends RichFlatMapFunction<Event, Event> {private ValueState<Boolean> seen;@Override public void open(OpenContext ctx) {seen = getRuntimeContext().getState(new ValueStateDescriptor<>("seen", Types.BOOLEAN));}@Override public void flatMap(Event e, Collector<Event> out) throws Exception {if (seen.value() == null) { // 首次出现out.collect(e);seen.update(true);}}
}

看到一个 ValueState<Boolean>，实际是分布式分片的键值存储：每个并行实例只持有自己 key 分片的那部分。

状态生命周期与清理

• 显式清理：seen.clear()（如 key 长期不活跃后）
• 状态 TTL：在 StateDescriptor 上配置，自动过期
• 复杂时序清理：结合 ProcessFunction + Timers（定时器回调里清状态）

6. Connected Streams：用“控制流”动态改变业务逻辑

典型用法：在线下发规则/阈值/黑白名单，实时影响主数据流的处理。

DataStream<String> control = env.fromData("DROP", "IGNORE").keyBy(x -> x);
DataStream<String> words   = env.fromData("Apache", "DROP", "Flink", "IGNORE").keyBy(x -> x);control.connect(words).flatMap(new ControlFunction()).print();public static class ControlFunction extends RichCoFlatMapFunction<String, String, String> {private ValueState<Boolean> blocked;@Override public void open(OpenContext ctx) {blocked = getRuntimeContext().getState(new ValueStateDescriptor<>("blocked", Boolean.class));}@Override public void flatMap1(String ctrl, Collector<String> out) { blocked.update(true); }@Override public void flatMap2(String word, Collector<String> out) throws Exception {if (blocked.value() == null) out.collect(word);}
}

关键：

• 两条流必须以一致的方式 keyBy，状态才可共享；
• 回调顺序不可控（flatMap1/2 谁先来由运行时决定），时序敏感就缓冲并自定调度。

7. 性能与工程化要点

序列化

• 优先 POJO/Avro；避免不稳定类型（随机、数组、枚举等）作为 key；
• 自定义类型须保证确定性的 hashCode/equals。

重分区

• 合理安排 keyBy，避免链路中多次无谓 shuffle；
• 利用 算子链 与 并行度 调优吞吐。

状态

• 预估 key 基数与状态大小；无界场景使用 TTL/窗口/清理；
• 大状态选 RocksDBStateBackend，权衡延迟与容量；
• 配置 Checkpoint（间隔/最小暂停/超时）与 外部化（HDFS/S3）。

一致性

• 端到端语义（at-least once/exactly once）取决于 source/sink 是否支持 2PC、幂等写等。

可观测性

• 关注反压、吞吐、延迟、状态大小、GC、checkpoint 时间；
• Web UI + 指标上报（Prometheus）+ 日志（JM/TM）。

8. 本地开发与调试

• IDE 断点：DataStream 支持本地调试，能 step into Flink 了解内部。
• 日志定位：JM/TM 日志、print() 输出前缀（1> / 2> 表示子任务）。
• 依赖打包：集群运行确保所有依赖在各节点可用（推荐 fat-jar/shade）。

9. 一个最小可运行的流式 ETL 骨架

需求：从 Source 读取行程 → NYC 清洗 → 富化网格 → 计算每个起点网格的最长行程 → 打印结果。

public class NycRideEtlJob {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<TaxiRide> rides = env.addSource(new TaxiRideSource(/* cfg */));DataStream<EnrichedRide> enriched = rides.flatMap(new NYCEnrichment()); // 过滤+富化DataStream<Tuple2<Integer, Integer>> longestByCell = enriched.flatMap((FlatMapFunction<EnrichedRide, Tuple2<Integer, Integer>>) (ride, out) -> {if (!ride.isStart) {int minutes = (int)((ride.endTime.getMillis() - ride.startTime.getMillis()) / 60000L);out.collect(Tuple2.of(ride.startCell, minutes));}}).keyBy(v -> v.f0) // startCell.maxBy(1);        // minuteslongestByCell.print();env.execute("NYC Ride ETL");}
}

上线前检查：checkpoint、重启策略、并行度、背压、告警通道、sink 幂等等。

10. 进一步学习与练习

• Hands-on：Rides & Fares、Ride Cleansing
• 专题：Stateful Stream Processing、DataStream Transformations
• 高级：ProcessFunction + Timers、Window/Watermark（事件时间）、ConnectedStreams 做流 join/规则推送、Async I/O 做维表富化