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

Flink Stream API 源码走读 - 总结

news 2025/8/17 11:35:31

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.dtcms.com/a/335036.html

相关文章：

  • 差分约束.
  • 腾讯混元大模型：实现3D打印产品生成的自动化平台
  • [Python 基础课程]继承
  • [Linux] RAID存储技术
  • 【102页PPT】电子行业数字化解决方案（附下载方式）
  • 容器化部署：用Docker封装机器翻译模型与服务详解
  • 服务器可以ping通，但部署的网站打不开
  • MyBatis 的 SQL 拦截器：原理、实现与实践
  • 基于Spring Boot的快递物流仓库管理系统 商品库存管理系统
  • OpenStack Neutron中的L2 Agent与L3 Agent：新手友好指南
  • Nginx蜘蛛请求智能分流：精准识别爬虫并转发SEO渲染服务
  • RemoteCtrl-初步的网络编程框架搭建
  • Linux 多线程：线程回收策略 线程间通信（互斥锁详解）
  • Easytier异地组网与Nginx反向代理
  • 昇腾AI自学Day2-- 深度学习基础工具与数学
  • 楼宇自控系统赋能建筑全维度管理，实现环境、安全与能耗全面监管
  • 计算分组内时间列的最大差值
  • 【AI论文】NextStep-1：迈向大规模连续令牌自回归图像生成
  • Warning: Unable to create personal MATLAB work folder:E:\绯荤粺榛樿\鏂囨。\MATLAB
  • 1083. 数列极差问题
  • 【深度学习】基于ESRNet模型的图像超分辨率训练
  • pytest介绍（python测试框架）（@pytest.mark.parametrize、@pytest.fixtures）
  • ClaudeCode使用指南
  • 鲁老师深度学习笔记（1）—最大似然估计
  • Flutter Provider 模式实现：基于 InheritedWidget 的状态管理实现
  • 93、23种设计模式之抽象工厂模式
  • 【读论文】医疗AI大模型：百川开源Baichuan-M2
  • 23. CommonJS 和 ES6 Module 区别
  • 19.3 Transformers量化模型极速加载指南：4倍推理加速+75%显存节省实战
  • ArrayList的contains问题