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

Flink Stream API 源码走读 - 总结

ops 2025/8/17 17:47:47

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.xdnf.cn/news/17941.html

相关文章：

  • 双指针和codetop复习
  • Day56 Java面向对象10 方法重写
  • Vue组件基础解析
  • [系统架构设计师]系统质量属性与架构评估（八）
  • Python语言---OrangePi全志H616
  • MySQL锁机制：悲观锁VS乐观锁详解
  • vector 手动实现 及遇到的各种细节问题
  • Azure AI Search 探索总结
  • 通配符 重定向 管道符
  • 数字分类：机器学习经典案例解析
  • vscode中使用CMake Tools生成compile_commands.json文件后，如何告诉clangd这个文件在哪里呢？
  • 【Linux系统】进程间通信：System V IPC——共享内存
  • 23. CommonJS 和 ES6 Module 区别
  • [1Prompt1Story] 生成行为控制器 | 语义向量重加权(SVR)
  • 【计算机视觉与深度学习实战】03基于Canny、Sobel和Laplacian算子的边缘检测系统设计与实现
  • Day11 栈与队列part2
  • duiLib 实现鼠标拖动状态栏时，窗口跟着拖动
  • webrtc弱网-VideoSendStreamImpl类源码分析与算法原理
  • 《Leetcode》-面试题-hot100-技巧
  • 嵌入式硬件篇---常见的单片机型号
  • 按键及消抖
  • Python环境下载安装、以及环境配置教程（Windows版）
  • java项目怎么实现用户行为分析、漏斗转化、数据可视化报表。
  • C语言零基础第18讲：自定义类型—结构体
  • 楼宇自控系统赋能建筑全维度管理，实现环境、安全与能耗全面监管
  • [Oracle数据库] Oracle 复杂查询
  • 当 GitHub 宕机时，我们如何协作？
  • Flink Sql 按分钟或日期统计数据量
  • 从 “视频孪生” 到 “视频动态目标三维重构”：技术演进与核心突破
  • PHP域名授权系统网站源码_授权管理工单系统_精美UI_附教程