【flink源码-1】【DataStream API】datastream程序各部分的作用、pipeline构建的意义、flink算子的内部结构有哪些之map源码分析

开始跟随课本张利兵的《Flink设计与实现核心原理与源码解析》阅读flink的源码希望对flink的流实现、窗口、容错、sql、jm与tm的通讯等逻辑与原理有一个更加细致的理解能够吸收flink的代码思想并运用到实际项目中已借鉴connector 类型转换的策略模式还有更重要的能够在遇到问题时知道出问题的地方源码大致逻辑是什么能比较效率的定位并解决问题。文章目录1. datastream的程序结构1.1. StreamExecutionEnvironment初始化1.2. 业务逻辑转换代码1.3. 执行应用程序2. DataStream的主要成员2.1. Pipeline拓扑的构建2.2. 算子的源码结构StreamTransformation结构StreamOperator用户自定义函数3. DataStream底层源码之map源码简介DataStream API主要用于构建流式类型的Flink应用处理实时无界数据流。DataStream API属于定义式编程接口可以编写复杂的例如对状态数据的操作、窗口的定义等功能。1. datastream的程序结构对于flink的datastream的执行程序来说程序固定的包含这几个动作StreamExecutionEnvironment初始化业务逻辑转换代码执行应用程序看一个代码的例子【wordcount】flink 代码快速入门具体的1.1. StreamExecutionEnvironment初始化主要通过DataStream API构建Flink作业需要的执行环境包括设定ExecutionConfig、CheckpointConfig等配置信息以及StateBackend和TimeCharacteristic等变量。1.2. 业务逻辑转换代码streamExecutionEnvironment提供了创建DataStream的方法例如socketTextStream(ip,port) 可以构建socket流。之后转换操作会以DataStreamSource为头部节点DataStream API中提供了各种转换操作例如map、reduce、join等算子用户可以用来构建完整的Flink计算逻辑。1.3. 执行应用程序编写完Flink应用后必须调用ExecutionEnvironment.execute()方法执行整个应用程序。execute()方法中会基于DataStream之间的转换操作生成StreamGraph并将StreamGraph结构转换为JobGraph最终将JobGraph提交到指定集群中运行。2. DataStream的主要成员DataStream通过转换操作可以生成新的DataStream。DataStream主要用于表达业务转换逻辑实际上并没有存储真实数据。2.1. Pipeline拓扑的构建StreamExecutionEnvironment会将DataStream之间的转换操作存储至其ListTransformation? transformations集合中然后基于这些转换操作构建作业Pipeline拓扑用于描述整个作业的计算逻辑。其中流式作业对应的Pipeline实现类为StreamGraph批作业对应的Pipeline实现类为Plan。2.2. 算子的源码结构StreamTransformation结构DataStream之间的转换操作都是通过StreamTransformation结构展示的。如上图当执行DataStream.map()方法转换时底层对应的便是OneInputTransformation转换操作。StreamOperator每个StreamTransformation都包含相应的StreamOperator。如上图执行DataStream.map-(new MapFunction(…))后会生成StreamMap算子。用户自定义函数StreamOperator包含了用户自定义函数的信息如上图StreamMap算子包含了MapFunction。MapFunction就是用户自定义的map转换函数。还有其他类型的函数例如ProcessFunction、SourceFunction和SinkFunction等。3. DataStream底层源码之map源码通过DataStream中的map转换操作为例对DataStream底层源码实现进行说明。map的使用逻辑.map(newMapFunctionString,Tuple2String,Integer(){publicTuple2String,Integermap(Stringword)throwsException{returnnewTuple2String,Integer(word,1);}}).用户首先定义MapFunction的具体逻辑然后调用DataStream.map() 将实例封装在map的转换中。接着看下map的源码/** 调用MapFunction来处理datastream的每一条数据。 */publicRSingleOutputStreamOperatorRmap(MapFunctionT,Rmapper,TypeInformationRoutputType){returntransform(Map,outputType,newStreamMap(clean(mapper)));}map调用了transform()进行后续处理会基于MapFunction实例创建StreamMap实例即StreamOperator的实现子类。transform()的具体逻辑在doTransform()中:protectedRSingleOutputStreamOperatorRdoTransform(StringoperatorName,TypeInformationRoutTypeInfo,StreamOperatorFactoryRoperatorFactory){//获取上一算子输出的typeinformation信息// read the output type of the input Transform to coax out errors about MissingTypeInfotransformation.getOutputType();//创建本算子的OneInputTransformationOneInputTransformationT,RresultTransformnewOneInputTransformation(this.transformation,operatorName,operatorFactory,outTypeInfo,environment.getParallelism());//主要用于每次转换操作后返回给用户继续操作的数据结构SuppressWarnings({unchecked,rawtypes})SingleOutputStreamOperatorRreturnStreamnewSingleOutputStreamOperator(environment,resultTransform);//将创建好的OneInputTransformation添加到StreamExecutionEnvironment的Transformation集合中用于生成StreamGraph对象。getExecutionEnvironment().addOperator(resultTransform);returnreturnStream;}逻辑如下1.获取TypeInformation从上一次转换操作中获取TypeInformation信息确定没有出现MissingTypeInfo错误以确保下游算子转换不会出现问题。2.创建OneInputTransformationOneInputTransformation也会包含当前DataStream对应的上一次转换操作。3.创建SingleOutputStreamOperatorSingleOutputStreamOperator继承了DataStream类属于特殊的DataStream主要用于每次转换操作后返回给用户继续操作的数据结构。4.将OneInputTransformation添加到trans集合调用addOperator(resultTransform)将创建好的OneInputTransformation添加到StreamExecutionEnvironment的Transformation集合中用于生成StreamGraph对象。5.按顺序存储将returnStream返回给用户继续执行后续的转换操作。基于这样连续的转换操作后会将所有DataStream的转换按顺序存储在StreamExecutionEnvironment中。小结在DataStream转换的过程中不管是哪种类型的转换操作都是按照同样的方式进行的首先将用户自定义的函数封装到Operator中然后将Operator封装到Transformation转换操作结构中最后将Transformation写入StreamExecutionEnvironment提供的Transformation集合。通过DataStream之间的转换操作形成Pipeline拓扑即StreamGraph数据结构最终通过StreamGraph生成JobGraph并提交到集群上运行。参考张利兵的《Flink设计与实现核心原理与源码解析》