一、背景Flink 提供了四层 API 抽象自底向上分别为┌─────────────────────────────────────────────────────────┐ │ SQL最高层声明式 │ ├─────────────────────────────────────────────────────────┤ │ Table API关系型操作 │ ├─────────────────────────────────────────────────────────┤ │ DataStream API流处理核心 API ← 本文重点 │ ├─────────────────────────────────────────────────────────┤ │ ProcessFunction最底层、最灵活 │ └─────────────────────────────────────────────────────────┘DataStream API 的定位在灵活性与易用性之间取得平衡——比 SQL/Table API 更底层、更灵活可直接操作事件时间、状态和定时器比 ProcessFunction 更具结构化提供了丰富的预定义算子。尽管 Flink 社区在大力推进 SQL/Table API 的功能覆盖但在以下场景中 DataStream API 仍不可替代复杂的自定义序列化/反序列化逻辑需要直接操作底层 State 后端如 MapState 的 TTL 精细控制异构多流的非等值关联超出 SQL JOIN 能力范围对延迟有极致要求的低延迟管道需要精细控制 buffer timeout二、核心原理与架构DataStream 执行架构图核心概念概念说明DataStream不可变的数据集合抽象支持有界/无界数据Transformation从一个或多个 DataStream 生成新 DataStream 的操作OperatorTransformation 的运行时实现Operator Chain相邻算子链接在同一线程内执行减少序列化/网络开销StreamGraph用户代码的逻辑执行计划JobGraph优化后的物理执行计划算子链合并ExecutionGraph并行化后的执行图每个算子对应多个并行实例Flink DataStream 程序采用懒执行机制调用 map()、filter() 等方法时并不会立即执行计算而是构建一个 DAG有向无环图。只有在显式调用 env.execute() 时才会将整个 DAG 提交给集群执行。这种设计的优势允许 Flink 对整个 DAG 做全局优化算子链接、资源分配支持在提交前进行计划验证三、程序骨架与执行模型每个 Flink DataStream 程序遵循以下五步模式// 1. 获取执行环境 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 配置执行参数 env.setParallelism(4); env.enableCheckpointing(60000); // 60s 一次 checkpoint // 3. 读取数据源 DataStreamString source env.addSource(new FlinkKafkaConsumer(...)); // 4. 定义转换逻辑 DataStreamResult result source .map(...) .keyBy(...) .window(...) .reduce(...); // 5. 输出结果 触发执行 result.addSink(new FlinkKafkaProducer(...)); env.execute(My Flink Job);执行流程图四、数据源 Source API从 Flink 1.12 起推荐使用新的 Source 接口统一批/流支持更好的扩展性旧的 SourceFunction 已被标记为过时。新接口将 Source 拆分为 SplitEnumerator分片枚举和 SourceReader数据读取两个组件。┌──────────────────────────────────────────────────────┐ │ Source (FLIP-27) │ ├──────────────────────┬───────────────────────────────┤ │ SplitEnumerator │ SourceReader │ │ (运行在 JobManager) │ (运行在 TaskManager) │ │ │ │ │ • 发现/分配分片 │ • 从分配的分片读取数据 │ │ • 协调多个 Reader │ • 维护分片级别的状态 │ │ • 支持动态发现新分片│ • 发送数据到下游算子 │ └──────────────────────┴───────────────────────────────┘常用预定义 Source// Kafka Source推荐使用新接口 KafkaSourceString kafkaSource KafkaSource.Stringbuilder() .setBootstrapServers(broker1:9092) .setTopics(input-topic) .setGroupId(my-group) .setStartingOffsets(OffsetsInitializer.earliest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStreamString stream env.fromSource( kafkaSource, WatermarkStrategy.noWatermarks(), Kafka Source ); // 文件 Source FileSourceString fileSource FileSource .forRecordStreamFormat(new TextLineInputFormat(), new Path(/data/input)) .monitorContinuously(Duration.ofSeconds(10)) .build(); // Collection Source用于测试 DataStreamInteger testStream env.fromElements(1, 2, 3, 4, 5);自定义Source示例// 基于新 Source 接口简化示意 public class MySource implements SourceMyEvent, MySplit, MyEnumState { Override public Boundedness getBoundedness() { return Boundedness.CONTINUOUS_UNBOUNDED; // 无界流 } Override public SplitEnumeratorMySplit, MyEnumState createEnumerator( SplitEnumeratorContextMySplit context) { return new MySplitEnumerator(context); } Override public SourceReaderMyEvent, MySplit createReader( SourceReaderContext context) { return new MySourceReader(context); } // ... 其他方法 }五、转换算子 TransformationFilter条件过滤// 过滤无效订单 DataStreamOrder validOrders orders.filter(order - order.getAmount() 0 order.getUserId() ! null );Map一对一转换适用场景数据清洗、格式转换、字段提取等一对一映射操作。// 将JSON字符串解析为POJO DataStreamOrder orders rawStream.map(new MapFunctionString, Order() { Override public Order map(String json) throws Exception { return objectMapper.readValue(json, Order.class); } }); // Lambda 简写 DataStreamInteger doubled intStream.map(value - value * 2);FlatMap一对多转换适用场景一条记录展开为多条如日志解析、订单拆分、条件过滤转换合一。// 将订单拆分为多个订单项事件 DataStreamOrderItem items orders.flatMap( new FlatMapFunctionOrder, OrderItem() { Override public void flatMap(Order order, CollectorOrderItem out) { for (OrderItem item : order.getItems()) { out.collect(item); } } } );KeyBy逻辑分区key 类型不能是数组POJO 作为 key 时必须正确重写 hashCode() 方法KeyBy 会触发网络 Shufflehash 分区// 按用户ID分区 KeyedStreamOrder, String keyedOrders orders.keyBy(Order::getUserId); // 使用元组字段 KeyedStreamTuple2String, Integer, String keyed tupleStream.keyBy(t - t.f0);Operator ChainingFlink 默认会将可以链接的算子合并在同一个 Task 线程中执行以减少线程切换和序列化开销。链接前Source → Map → Filter → KeyBy → Window → Sink ↓ (network shuffle) 链接后[Source → Map → Filter] → [KeyBy → Window → Sink] (一个 Operator Chain) (另一个 Operator Chain)可通过以下方式控制链接行为// 全局禁用算子链 env.disableOperatorChaining(); // 单个算子断开前向链 stream.map(...).startNewChain(); // 单个算子完全禁用链接 stream.map(...).disableChaining();六、数据汇 Sink API与 Source 类似Sink 也经历了接口重构。从1.14版本开始使用新的Sink接口新统一接口支持精确一次语义旧的SinkFunction接口被标记为过时。常用 Sink 示例// Kafka Sink新接口 KafkaSinkString kafkaSink KafkaSink.Stringbuilder() .setBootstrapServers(broker1:9092) .setRecordSerializer( KafkaRecordSerializationSchema.builder() .setTopic(output-topic) .setValueSerializationSchema(new SimpleStringSchema()) .build() ) .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // 精确一次 .setTransactionalIdPrefix(my-app) .build(); stream.sinkTo(kafkaSink); // FileSink精确一次 文件滚动 FileSinkString fileSink FileSink .forRowFormat(new Path(/output), new SimpleStringEncoderString(UTF-8)) .withRollingPolicy( DefaultRollingPolicy.builder() .withRolloverInterval(Duration.ofMinutes(15)) .withInactivityInterval(Duration.ofMinutes(5)) .withMaxPartSize(MemorySize.ofMebiBytes(1024)) .build() ) .build(); stream.sinkTo(fileSink);writeAsText() / print() 等便捷方法不参与 Checkpoint仅适合调试不保证 Exactly-Once。生产环境务必使用 sinkTo() 或 addSink() 结合支持事务的 Connector。