Flink任务如何跑起来之 1.DataStream和Transformation

Flink任务如何跑起来之 1.DataStream和Transformation

1. 滥觞

在使用Flink完成业务功能之余,有必要了解下我们的任务是如何跑起来的。知其然,知其所以然。

既然重点是学习应用程序如何跑起来,那么应用程序的内容不重要,越简单越好。
WordCount示例作为学习数据引擎时hello word程序,再合适不过。接下来便以任务执行顺序为线索开启对源码逐步学习。

public class WordCount {public static void main(String[] args) throws Exception {// 初始化执行环境Configuration configuration = new Configuration();configuration.setString("rest.port", "9091");StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);env.setParallelism(1);// 业务逻辑转换DataStream<String> text = env.fromCollection(Arrays.asList("zhangsan", "lisi", "wangwu", "zhangsan")).name("zl-source");DataStream<Tuple2<String, Integer>> counts = text.map(row -> Tuple2.of(row, 1)).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(value -> value.f0).sum(1).name("counter");counts.print().name("print-sink");// 执行应用程序env.execute("WordCount");}
}

为了使示例代码足够纯粹(直接复制粘贴后即可跑起来的那种),因此在示例中直接使用List数据作为Source。

最后,计划将自己学习的过程以系列文档的形式作为记录。同时作为自己学习过程的记录,可能存在错误或片面理解,欢迎一起讨论。

2. 头疼的“角色”

在学习源码或查阅资料的同时,以下单词(但不限于)一定会频繁出现,它们或者直接对应flink源码中的接口、类名,或者是一些概念名称。初次看到难免让人抓狂。现在先对这些单词混个脸熟。

Client
JobManager/JobMaster
TaskManager/TaskExecutor
Transformation
StreamOperator
StreamGraph
JobGraph
ExecutionGraph
Task
StreamTask
……

3. 宏观视角

当任务开始执行后,便可以在WebUI上查看其对应的物理执行拓扑,即Task DAG。从我们编写的应用程序代码到Task DAG势必经历了复杂的解析转换操作,这个过程大体如下所示。

在这里插入图片描述

我们编写的应用程序代码首先会转化为Transformation,该实例将作为Flink世界中的起点,开启了之后一系列“旅程”。

4. env.execute()方法做了什么?

在使用DataStream API编写应用程序时,无论业务逻辑如何如何的复杂,但整体结构大致由三部分构成,即

// 1.初始化执行环境
StreamExecutionEnvironment env = ;
// 2.业务逻辑转换,即一系列的DataStream转化
DataStream source = ;
// 3.env.execute()
env.execute();

既然最后必须执行 env.execute()方法,那么首先了解下execute都执行了那些操作。

基于1.16版本的源码,并只保留了源码中的关键逻辑。

// 方法1
public JobExecutionResult execute(String jobName) throws Exception {final List<Transformation<?>> originalTransformations = new ArrayList<>(transformations);// 生成StreamGraph,最终调用方法4,通过StreamGraphGenerator生成StreamGraphStreamGraph streamGraph = getStreamGraph();// ...try {// 调用方法2return execute(streamGraph);} catch (Throwable t) {// ...}
}
// 方法2
public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {// 调用方法3,通过StreamGraph最终得到JobClientfinal JobClient jobClient = executeAsync(streamGraph);try {final JobExecutionResult jobExecutionResult;// ...jobListeners.forEach(jobListener -> jobListener.onJobExecuted(jobExecutionResult, null));return jobExecutionResult;} catch (Throwable t) {// ...}
}
// 方法3
public JobClient executeAsync(StreamGraph streamGraph) throws Exception {// 根据启动环境,得到对应环境的Executor实现// 如miniCluster环境则对应LocalExecutorfinal PipelineExecutor executor = getPipelineExecutor();// 在具体的executor.execute方法中,将StreamGraph先转化成JobGraph,在将JobGraph提交到JobManager中CompletableFuture<JobClient> jobClientFuture =executor.execute(streamGraph, configuration, userClassloader);try {JobClient jobClient = jobClientFuture.get();jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(jobClient, null));collectIterators.forEach(iterator -> iterator.setJobClient(jobClient));collectIterators.clear();return jobClient;} catch (ExecutionException executionException) {// ...}
}
// 方法4
private StreamGraph getStreamGraph(List<Transformation<?>> transformations) {synchronizeClusterDatasetStatus();// 根据Transformation生成StreamGraphreturn getStreamGraphGenerator(transformations).generate();
}

通过上述源码调用链可以,完成从DataStream API->Transformation->StreamGraph->JobGraph的转化。最后将JobGraph提交到了JobManager中并,执行后续操作。

从上述方法4getStreamGraph(List<Transformation<?>> transformations)可知,StreamGraph由Transformation演变而来,此处不禁会产生一个新的疑问,Transformation又从何而来?

WordCount示例代码中并没有与Transformation直接相关的代码。通过查看getSreamGraph方法的完成调用链可知其入参直接来自是StreamExecutionEnvironment类中的transformations成员属性值。在应用程序第一步便生成了StreamExecutionEnvironment的实例,接下来通过env得到DataStream并进行了一系列的转化操作,而在最后的execute方法中便已直接使用transformations属性值了,那么该属性中一定是前面2个过程中实际赋值的。

protected final List<Transformation<?>> transformations = new ArrayList<>();

5. Transformation何时生成?

从StreamExecutionEnvironment的源码中可知,transformations属性只有addOperator方法会执行集合的add操作,其余地方均为集合的get操作。
然而addOperator方法有诸多调用方,且均为其他类中的调用,继续往上查看调用方有些困难,因此这里暂时记下addOperator方法唯一对transformations集合中执行add操作的结论。

// 该方法不适合用户使用。创建operator的api方法必须调用此方法
@Internal
public void addOperator(Transformation<?> transformation) {Preconditions.checkNotNull(transformation, "transformation must not be null.");this.transformations.add(transformation);
}

通过查看StreamExecutionEnvironment实例的创建过程,可以发现在创建过程中并无transformations的add操作,因此是在DataStream转换操作中对transformations执行了add操作。

5.1. DataStream

在Flink中使用DataStream表示数据流。其仅用于表达业务转化逻辑,实际上并没有真正的存储数据。

DataSteam是顶层封装类,其子类如下

在这里插入图片描述
DataStream类中只有两个成员属性,分别是StreamExecutionEnvironment和Transformation,并在构造方法中对其进行初始化。因此实例化DataStream的同时除执行环境外,还必须传入Transformation的实例。

public class DataStream<T> {protected final StreamExecutionEnvironment environment;protected final Transformation<T> transformation;public DataStream(StreamExecutionEnvironment environment, Transformation<T> transformation) {this.environment =Preconditions.checkNotNull(environment, "Execution Environment must not be null.");this.transformation =Preconditions.checkNotNull(transformation, "Stream Transformation must not be null.");}// ...
}

回到WordCount示例代码中,从集合到DataStream的过程,封装示意如下。

在这里插入图片描述

注意,Transformation中并不是直接持有了AbstractUdfStreamOperator的引用,而是对应的工厂。

源码中关键步骤如下

// 步骤1,从List到Function
public <OUT> DataStreamSource<OUT> fromCollection(Collection<OUT> data, TypeInformation<OUT> typeInfo) {// ...// 创建SourceFunction实例,SourceFunction是Function的实现SourceFunction<OUT> function = new FromElementsFunction<>(data);return addSource(function, "Collection Source", typeInfo, Boundedness.BOUNDED).setParallelism(1);
}// 步骤2,从Function到StreamOperator
private <OUT> DataStreamSource<OUT> addSource(final SourceFunction<OUT> function,final String sourceName,@Nullable final TypeInformation<OUT> typeInfo,final Boundedness boundedness) {// ...// 创建StreamSource实例,StreamSource是AbstractUdfStreamOperator的子类,Flink中算子的表示final StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function);return new DataStreamSource<>(this, resolvedTypeInfo, sourceOperator, isParallel, sourceName, boundedness);
}// 步骤3,从StreamOperator到Transformation,再到DataStream
public DataStreamSource(StreamExecutionEnvironment environment,TypeInformation<T> outTypeInfo,StreamSource<T, ?> operator,boolean isParallel,String sourceName,Boundedness boundedness) {super(environment,// 创建Transformation实例,Transformation是PhysicalTransformation的子类new LegacySourceTransformation<>(sourceName,// 将StreamSource封装到Transformation中operator,outTypeInfo,environment.getParallelism(),boundedness));// ...
}

继续查看DataStream的map操作可以可以发现,核心流程和上述由集合创建DataStream的过程基本一致:

  • 首先创建Function实例
  • 其次由Function实例创建AbstractUdfStreamOperator实例
  • 然后将AbstractUdfStreamOperator实例封装到Transformation实例中
  • 最后由Transformation和StreamExecutionEnvironment实例创建DataStream实例

不同之处在于,map操作最后将得到的PhysicalTransformation实例添加到StreamExecutionEnvironment实例中的transformations集合中去了。这点差异其实和Transformation实例表示的含义有关,放在文章末尾解释。

protected <R> SingleOutputStreamOperator<R> doTransform(String operatorName,TypeInformation<R> outTypeInfo,StreamOperatorFactory<R> operatorFactory) {// ...OneInputTransformation<T, R> resultTransform =new OneInputTransformation<>(this.transformation,operatorName,operatorFactory,outTypeInfo,environment.getParallelism());SingleOutputStreamOperator<R> returnStream =new SingleOutputStreamOperator(environment, resultTransform);// 区别:添加Transformation到StreamExecutionEnvironment中getExecutionEnvironment().addOperator(resultTransform);return returnStream;
}

但并不是全部的DataStream转化操作都需要经历上述将Function实例封装成AbstractUdfStreamOperator实例,然后将AbstractUdfStreamOperator实例封装到PhysicalTransformation实例的过程。如示例代码中的keyBy和sum操作。其中keyBy并未直接涉及Function,而sum操作直接将得到的SumAggregator函数实例封装到了ReduceTransformation实例中,然后由ReduceTransformation实例得到DataStream实例。

5.2. Transformation

DataStream面向开发者,而Transformation面向flink内核。
每个DataStream实例中都包含一个Transformation实例,表示当前Datastream从上游的DataStream使用该Transformation而来。而所有DataStream中Transformation又都添加到了StreamExecutionEnvironment实例中的transformations集合中去,用于接下来的StreamGraph实例的生成。
Transformation中记录了上游的数据来源,但其并关心数据的物理来源、序列化、转发等问题。

Transformatio是顶层抽象类,有众多的子类,涵盖了DataStream的所有转换,其直接子类如下,可以分为两大类

  • PhysicalTransformation,将会转换成后续graph中节点信息
  • 非PhysicalTransformation,将会转换成后续graph中的边信息

在这里插入图片描述
Transformation中属性如下所示,其中Optional<SlotSharingGroup>表示共享槽位信息,只有开启了允许共享槽位后,该属性才会被设置值。

其构造方法如下,除name外还需要输出类型和并行度两个参数。

public Transformation(String name, TypeInformation<T> outputType, int parallelism) {this.id = getNewNodeId();this.name = Preconditions.checkNotNull(name);this.outputType = outputType;this.parallelism = parallelism;this.slotSharingGroup = Optional.empty();
}

PhysicalTransformation仅在其父类的基础上增加了设置ChainingStrategy的方法,用于表示生成算子链的策略。

@Internal
public abstract class PhysicalTransformation<T> extends Transformation<T> {PhysicalTransformation(String name, TypeInformation<T> outputType, int parallelism) {super(name, outputType, parallelism);}/** Sets the chaining strategy of this {@code Transformation}. */public abstract void setChainingStrategy(ChainingStrategy strategy);
}

PhysicalTransformation中有众多的实现子类,全部子类继承关系如下。

在这里插入图片描述
其中以下几个子类出场频率相对更高一些,其他子类只有我们的业务逻辑比较复杂时才会用到。

  • LegacySourceTransformation 表示Source的Transformation
  • LegacySinkTransformation 表示Sink的Transformation
  • SourceTransformation
  • SinkTransformation
  • OneInputTransformation 表示单个输入流的Transformation,如常见的map、flatMap、fliter等
  • TwoInputTransformation 表示两个输入流的Transformation,如concat

疑问:为什么Source和Sink都各自分别有两个Transformation子类?
通过名称也可以看出一些端倪,新老两种实现。
在1.14版本之前,分别通过env.addSource(SourceFunction)DataStream.addSink(SinkFunction)方法生成source和sink
从1.14版本开始新增了env.fromSource(Source)DataStream.sinkTo(Sink)的方式生成source和sink。
新旧方法中入参类型不同,因此导致了两种不同的Transformation实现,从各自的实现类中也可以体现这一点,如下所示。

public class LegacySourceTransformation<T> extends PhysicalTransformation<T>implements WithBoundedness {// sourceFunction的引用private final StreamOperatorFactory<T> operatorFactory;// ...
}public class SourceTransformation<OUT, SplitT extends SourceSplit, EnumChkT>extends PhysicalTransformation<OUT> implements WithBoundedness {// source的引用private final Source<OUT, SplitT, EnumChkT> source;// ...
}public class LegacySinkTransformation<T> extends PhysicalTransformation<T> {private final Transformation<T> input;// sinkFunction的引用private final StreamOperatorFactory<Object> operatorFactory;// ...
}public class SinkTransformation<InputT, OutputT> extends PhysicalTransformation<OutputT> {private final DataStream<InputT> inputStream;// sink的引用private final Sink<InputT> sink;private final Transformation<InputT> input;// ...
}

Source作为整个数据流的头部,不存在上游,因此其Transformation实现中没有上游Transformation的引用,除此之外其余的Transformation子类中,均持有一个表示上游Transformation的引用,如上述sink中的input属性。

最后解释下,前面提到的为什么没有将表示Source的DataStream中的Transformation加入到env中表示Transformation的集合中,而接下来的转化中,将对应的Transformation加入到了env中。因为Source作为数据源的头部,不会存在上游,而Source作为其他DataSteam的上游,一定会加入到其Transformation的input中,因此没必要单独将Source的transformation加入到env中。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/pingmian/23508.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

好家风短视频:成都鼎茂宏升文化传媒公司

好家风短视频&#xff1a;传承与发扬家庭美德的新载体 在数字时代的浪潮中&#xff0c;短视频以其简短、生动、直观的特点&#xff0c;成为了人们获取信息、传递情感的重要渠道。成都鼎茂宏升文化传媒公司而在这个多元化的内容生态中&#xff0c;好家风短视频以其独特的价值和…

5.透明效果

实时渲染中要实现透明效果&#xff0c;通常会在渲染模型时控制它的透明通道&#xff08;Alpha channel&#xff09;。 当一个物体被渲染到屏幕上时&#xff0c;每个片元除了颜色和深度值之外&#xff0c;它还有另一个属性—透明度。 当透明度为1时&#xff0c;表示该像素是完…

Dvws靶场

文章目录 一、XXE外部实体注入二、No-SQL注入三、Insecure Direct Object Reference四、Mass Assignment五、Information Disclosure六、Command Injection七、SQL注入 一、XXE外部实体注入 访问http://192.168.92.6/dvwsuserservice?wsdl&#xff0c;发现一个SOAP服务。在SO…

MySQL 存储过程(二)

本篇继续介绍MySQL存储过程的相关内容。 目录 一、if语句 二、case 三、循环语句 while loop repeat 一、if语句 在存储过程中&#xff0c;可以使用if语句进行条件判断&#xff0c;其语法结构如下&#xff1a; if 判断语句 then 逻辑语句..... elseif 判断语句 then 逻…

【Linux取经路】初识信号

文章目录 一、人眼中的信号 VS 进程眼中的信号二、ctrlc 终止一个前台进程三、查看信号信息3.1 Core dump——核心转储功能验证 四、信号的处理方式五、ctrlc 被解释成2号信号验证5.1 signal——设置自定义捕捉方法 六、ctrlc 是如何变成信号的&#xff1f;七、异步、软中断八、…

【SQL边干边学系列】04中级问题(续)

文章目录 前言回顾中级问题25.高昂运费26.2015年的高昂运费27.高昂运维 - 使用between28.去年的高昂运费29.库存清单30.没有任何订单的客户31.没有任何订单的客户&#xff0c;员工ID为4 答案25.高昂运费26.2015年的高昂运费27.高昂运维 - 使用between28.去年的高昂运费29.库存清…

CVE-2022-4230

CVE-2022-4230 漏洞介绍 WP Statistics WordPress 插件13.2.9之前的版本不会转义参数&#xff0c;这可能允许经过身份验证的用户执行 SQL 注入攻击。默认情况下&#xff0c;具有管理选项功能 (admin) 的用户可以使用受影响的功能&#xff0c;但是该插件有一个设置允许低权限用…

DDMA信号处理以及数据处理的流程---DDMA原理介绍

Hello&#xff0c;大家好&#xff0c;我是Xiaojie&#xff0c;好久不见&#xff0c;欢迎大家能够和Xiaojie一起学习毫米波雷达知识&#xff0c;Xiaojie准备连载一个系列的文章—DDMA信号处理以及数据处理的流程&#xff0c;本系列文章将从目标生成、信号仿真、测距、测速、cfar…

vscode设置代码自动换行显示

☆ 问题描述 vscode设置代码自动换行显示 ★ 解决方案 ✅ 总结

每日两题 / 198. 打家劫舍 74. 搜索二维矩阵(LeetCode热题100)

198. 打家劫舍 - 力扣&#xff08;LeetCode&#xff09; dp[i]表示考虑前i 1号房屋&#xff0c;能获取的最大金额。对于没一间房屋都有偷与不偷两种选择 如果偷&#xff0c;需要从dp[i - 2]转移&#xff0c;因为不能偷窃相邻房屋&#xff0c;dp[i] dp[i - 2] nums[i] 如果…

稍微学学react

文章开始前&#xff0c;先划划水~ 今日份开心&#xff1a; 今天看之前发布的按钮npm包下载量有162次&#xff0c;早知道好好做了 今日份不开心&#xff1a; 爬岗位看到一个整体都挺满意的岗位&#xff0c;公司位置和发展大方向都好喜欢&#xff01;&#xff01;&#xff01;…

【PL理论】(6) F#:标准库之列表(List)

​​​​​ &#x1f4ad; 写在前面&#xff1a;本章我们将介绍 F# 标准库的列表&#xff0c;我们将简单的先过一遍列表的一些常用操作&#xff0c;具体的讲解我们将放在后续章节。 目录 0x00 标准库&#xff1a;列表&#xff08;List&#xff09; 0x01 模式匹配与列表 0x…

14.8k Star!CrewAI:部署一支由你指挥的人工智能代理大军,股票分析、发布帖子、支持Ollama!

原文链接&#xff1a;&#xff08;更好排版、视频播放、社群交流、最新AI开源项目、AI工具分享都在这个公众号&#xff01;&#xff09; 14.8k Star&#xff01;CrewAI&#xff1a;部署一支由你指挥的人工智能代理大军&#xff0c;股票分析、发布帖子、支持Ollama&#xff01;…

计算机组成结构—多处理器

目录 一、SISD、SIMD、MIMD 和向量处理器 1. 费林分类法 2. SIMD 和向量处理器 二、硬件多线程 三、多核处理器和 SMP 1. 多核处理器 2. 共享内存多处理器&#xff08;SMP&#xff09; 3. MPP 和集群 一、SISD、SIMD、MIMD 和向量处理器 通过改进系统结构&#xff0c;可…

C++设计模式-外观模式,游戏引擎管理多个子系统,反汇编

运行在VS2022&#xff0c;x86&#xff0c;Debug下。 30. 外观模式 为子系统定义一组统一的接口&#xff0c;这个高级接口会让子系统更容易被使用。应用&#xff1a;如在游戏开发中&#xff0c;游戏引擎包含多个子系统&#xff0c;如物理、渲染、粒子、UI、音频等。可以使用外观…

图像算法---自动曝光AE

一&#xff0c;自动曝光AE 自动曝光&#xff08;AE&#xff0c;全称Auto Exposure&#xff09;是一种在摄影和摄像中广泛使用的技术&#xff0c;它允许相机或摄像机根据环境光线条件自动调整曝光参数&#xff0c;以获得清晰、亮度适中的图像或视频。以下是关于自动曝光AE的详细…

31|HTTP3:甩掉TCP、TLS 的包袱,构建高效网络

前面两篇文章我们分析了HTTP/1和HTTP/2&#xff0c;在HTTP/2出现之前&#xff0c;开发者需要采取很多变通的方式来解决HTTP/1所存在的问题&#xff0c;不过HTTP/2在2018年就开始得到了大规模的应用&#xff0c;HTTP/1中存在的一大堆缺陷都得到了解决。 HTTP/2的一个核心特性是…

优卡集团冲刺港股上市:90后创始团队孵化,IPO前突击大额分红

现年26岁的鲁圳&#xff0c;正在带领其6年以来的创业成果冲击资本市场。 近日&#xff0c;金融居间机构服务商优卡集团&#xff08;Yoc Group&#xff09;向港交所递交上市申请&#xff0c;民银资本为其独家保荐人。透过招股书可知&#xff0c;优卡集团成立于2018年&#xff0…

【CTF-Web】文件上传漏洞学习笔记(ctfshow题目)

文件上传 文章目录 文件上传What is Upload-File&#xff1f;Upload-File In CTFWeb151考点&#xff1a;前端校验解题&#xff1a; Web152考点&#xff1a;后端校验要严密解题&#xff1a; Web153考点&#xff1a;后端校验 配置文件介绍解题&#xff1a; Web154考点&#xff1a…

vcruntime140.dll找不到的正确处理方法,vcruntime140.dll是什么文件

vcruntime140.dll找不到的这个问题&#xff0c;相信不少人都有遇到吧&#xff1f;其实遇到了也不需要害怕&#xff0c;这只是一个很小的问题&#xff0c;我们完全可以自己快速修复。还有就是缺失这个vcruntime140.dll文件&#xff0c;你的一些程序是会打不开的&#xff0c;不用…