当前位置： 首页 > news >正文

Flink Stream API 源码走读 - 总结

news 2025/8/17 16:23:39

Flink Stream API 源码走读 - 总结

概述

本文档详细分析Flink数据流处理的完整流程，从API调用到Transformation构建，再到最终的执行图生成。通过源码分析，深入理解Flink如何将用户的流式API调用转换为可执行的数据流图。
流程概览

1. Source阶段：数据源的创建

1.1 API调用

// 用户代码
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> source = env.socketTextStream("localhost", 9999);

1.2 内部处理流程

当调用env.socketTextStream()时，Flink内部会：

  1. 创建SourceFunction：

    • 创建SocketTextStreamFunction实例
    • 封装Socket连接逻辑和数据读取逻辑

  2. 生成LegacySourceTransformation：

    // 源码位置：LegacySourceTransformation.java
public class LegacySourceTransformation<T> extends PhysicalTransformation<T> {private final StreamOperatorFactory<T> operatorFactory;// ...
}
    • 包含具体的运算逻辑（SocketTextStreamFunction）
    • 通过SimpleOperatorFactory.of(operator)包装成算子工厂
    • 这是一个物理Transformation，包含实际的计算逻辑

  3. 返回DataStreamSource：

    // DataStreamSource继承自SingleOutputStreamOperator
public class DataStreamSource<T> extends SingleOutputStreamOperator<T> {// this.transformation 引用上面创建的LegacySourceTransformation
}

2. Map阶段：数据转换操作

2.1 API调用

// 用户代码
SingleOutputStreamOperator<String> mapped = source.map(new MapFunction<String, String>() {@Overridepublic String map(String value) throws Exception {return value.toUpperCase();}
});

2.2 内部处理流程

当调用map()方法时：

  1. 类型推断：

    // DataStream.java 第588行
public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper) {TypeInformation<R> outType = TypeExtractor.getMapReturnTypes(clean(mapper), getType(), Utils.getCallLocationName(), true);return map(mapper, outType);
}

  2. 创建StreamMap算子：

    // DataStream.java 第631行
return transform("Map", outputType, new StreamMap<>(clean(mapper)));

  3. 生成OneInputTransformation：

    // DataStream.doTransform() 方法
OneInputTransformation<T, R> resultTransform = new OneInputTransformation<>(this.transformation,    // 引用上一个transformation（LegacySourceTransformation）operatorName,          // "Map"operatorFactory,       // SimpleOperatorFactory.of(StreamMap)outTypeInfo,          // 输出类型信息environment.getParallelism(),false
);

  4. 添加到执行环境：

    // 将物理transformation添加到transformations列表
getExecutionEnvironment().addOperator(resultTransform);

  5. 返回新的DataStream：

    SingleOutputStreamOperator<R> returnStream =new SingleOutputStreamOperator(environment, resultTransform);

3. FlatMap阶段：扁平化映射操作

3.1 API调用

// 用户代码
SingleOutputStreamOperator<String> flatMapped = mapped.flatMap(new FlatMapFunction<String, String>() {@Overridepublic void flatMap(String value, Collector<String> out) throws Exception {String[] words = value.split(" ");for (String word : words) {out.collect(word);}}
});

3.2 内部处理流程

FlatMap操作的处理流程与Map类似：

  1. 创建StreamFlatMap算子：包装用户的FlatMapFunction
  2. 生成OneInputTransformation：引用上一个Map的transformation
  3. 添加到transformations列表：这是物理transformation
  4. 返回新的SingleOutputStreamOperator

4. KeyBy阶段：数据分组操作

4.1 API调用

// 用户代码
KeyedStream<String, String> keyed = flatMapped.keyBy(value -> value);

4.2 内部处理流程

KeyBy操作比较特殊：

  1. 生成PartitionTransformation：

    // 源码位置：PartitionTransformation.java
public class PartitionTransformation<T> extends Transformation<T> {private final Transformation<T> input;  // 引用上一个transformationprivate final StreamPartitioner<T> partitioner;  // 分区器，不是算子工厂
}

  2. 关键特点：

    • 这是一个虚拟transformation，不包含实际的计算逻辑
    • 持有的是KeyGroupStreamPartitioner分区器，而不是算子工厂
    • 不会添加到transformations列表中
    • 用于指导数据在网络中的分发策略

  3. 返回KeyedStream：

    public class KeyedStream<T, KEY> extends DataStream<T> {// 继承自DataStream，但提供了按键分组后的特殊API
}

5. Window阶段：窗口操作

5.1 API调用

// 用户代码
WindowedStream<String, String, TimeWindow> windowed =keyed.timeWindow(Time.seconds(5));

5.2 内部处理流程

Window操作创建了一个特殊的中间对象：

  1. 创建WindowedStream：

    // 源码位置：WindowedStream.java 第79行
public WindowedStream(KeyedStream<T, K> input, WindowAssigner<? super T, W> windowAssigner) {this.input = input;  // 保持对上游KeyedStream的引用this.builder = new WindowOperatorBuilder<>(windowAssigner,                    // 窗口分配器windowAssigner.getDefaultTrigger(input.getExecutionEnvironment()),  // 触发器input.getExecutionConfig(),        // 执行配置input.getType(),                   // 输入类型input.getKeySelector(),            // Key选择器input.getKeyType()                 // Key类型);
}

  2. WindowedStream特点：

    • 不是真正的DataStream，只是一个工具类/API容器
    • 不能调用map、flatMap等通用流操作
    • 只能调用窗口相关的API：reduce、apply、process等
    • 持有WindowOperatorBuilder，用于后续构建窗口算子

6. Reduce阶段：窗口聚合操作

6.1 API调用

// 用户代码
SingleOutputStreamOperator<String> result = windowed.reduce(new ReduceFunction<String>() {@Overridepublic String reduce(String value1, String value2) throws Exception {return value1 + "," + value2;}
});

6.2 内部处理流程

这是窗口操作的核心实现：

  1. 构建WindowOperator：

    // WindowedStream.java 第230行
OneInputStreamOperator<T, R> operator = builder.reduce(reduceFunction, function);

  2. WindowOperatorBuilder.reduce()方法：

    • 创建WindowOperator实例
    • 将用户的ReduceFunction和内部的PassThroughWindowFunction组合
    • WindowOperator包含完整的窗口处理逻辑：
      • 窗口分配（WindowAssigner）
      • 窗口触发（Trigger）
      • 数据聚合（ReduceFunction）
      • 结果输出（WindowFunction）

  3. 生成OneInputTransformation：

    // WindowedStream.java 第233行
return input.transform(opName, resultType, operator).setDescription(opDescription);
    • 这是一个物理transformation
    • 包含具体的窗口+聚合运算逻辑
    • 会被添加到transformations列表中

  4. 返回SingleOutputStreamOperator：

    • 又回到了正常的DataStream
    • 可以继续进行其他流操作

7. Sink阶段：数据输出操作

7.1 API调用

// 用户代码
DataStreamSink<String> sink = result.print();

7.2 内部处理流程

Sink操作是数据流的终点：

  1. 创建PrintSinkFunction：

    // 内部会创建PrintSinkFunction实例
// 包含将数据打印到控制台的逻辑

  2. 生成LegacySinkTransformation：

    // 通过SimpleOperatorFactory.of()包装成算子工厂
// 创建LegacySinkTransformation实例
LegacySinkTransformation<T> sinkTransformation = new LegacySinkTransformation<>(input.getTransformation(),  // 引用上一个transformationname,                       // "Sink: Print to Std. Out"operatorFactory,           // SimpleOperatorFactory.of(StreamSink)parallelism
);

  3. 添加到transformations列表：

    • 这是一个物理transformation
    • 包含具体的数据处理逻辑（打印到屏幕）
    • 会被添加到环境的transformations列表中

  4. 返回DataStreamSink：

    public class DataStreamSink<T> {private final Transformation<T> transformation;// 不是真正的流，只是一个封装工具类// 不能再调用map、flatMap等API// 只持有自己的transformation引用
}

8. Transformation链式结构分析

8.1 物理Transformation vs 虚拟Transformation

物理Transformation（会添加到transformations列表）：

  • LegacySourceTransformation：包含Source算子逻辑
  • OneInputTransformation：包含Map、FlatMap、Window等算子逻辑
  • LegacySinkTransformation：包含Sink算子逻辑

虚拟Transformation（不会添加到transformations列表）：

  • PartitionTransformation：只包含分区策略，用于指导数据分发

8.2 引用关系链

LegacySourceTransformation (Source)↑ input
OneInputTransformation (Map)↑ input
OneInputTransformation (FlatMap)↑ input
PartitionTransformation (KeyBy) [虚拟]↑ input
OneInputTransformation (Window+Reduce)↑ input
LegacySinkTransformation (Sink)

8.3 关键特点

  1. Source Transformation：

    • 没有input引用（数据流的起点）
    • 包含数据源逻辑

  2. 中间Transformation：

    • 都有input成员变量，引用上一个transformation
    • 形成链式结构

  3. Sink Transformation：

    • 有input引用，但没有下游
    • 数据流的终点

  4. 遍历能力：

    • 通过最后一个transformation可以向上遍历整个链路
    • 这是Flink构建执行图的基础

返回目录

Flink 源码系列 - 前言

http://www.lryc.cn/news/623298.html

相关文章：

  • 楼宇自控系统赋能建筑全维度管理，实现环境、安全与能耗全面监管
  • STM32硬件SPI配置为全双工模式下不要单独使用HAL_SPI_Transmit API及HAL_SPI_TransmitReceive改造方法
  • 【时时三省】(C语言基础)共用体类型数据的特点
  • Langfuse2.60.3:独立数据库+docker部署及环境变量详细说明
  • Java 中重载与重写的全面解析（更新版）
  • Mybatis-3自己实现MyBatis底层机制
  • 从冒泡到快速排序：探索经典排序算法的奥秘（二）
  • PHP反序列化的CTF题目环境和做题复现第1集
  • 企业运维规划及Linux介绍虚拟环境搭建
  • python---包
  • 一文速通Python并行计算：14 Python异步编程-协程的管理和调度
  • CF每日3题(1500-1700)
  • P2169 正则表达式
  • w嵌入式分享合集66
  • 【Bluedroid】A2DP控制通道UIPC机制深度解析（btif_a2dp_control_init）
  • Java8~Java21重要新特性
  • JAVA面试汇总（四）JVM（一）
  • 028 动静态库 —— 动态库
  • duiLib 实现鼠标拖动标题栏时，窗口跟着拖动
  • Vue 3.5重磅更新：响应式Props解构，让组件开发更简洁高效
  • 分享一个Oracle表空间自动扩容与清理脚本
  • CPP多线程3：async和future、promise
  • MATLAB基础训练实验
  • 超越“调参”：从系统架构师视角，重构 AI 智能体的设计范式
  • 深度剖析Redisson分布式锁项目实战
  • 【数据分享】大清河（大庆河）流域上游土地利用
  • AutoDL使用学习
  • K8s核心组件全解析
  • 服务器配置开机自启动服务
  • GEEPython-demo1：利用Sentinel-2监测北京奥林匹克森林公园2024年NDVI变化(附Python版)