Flink流式计算从入门到实战 二
文章目录
- 三、Flink运行架构
- 1、JobManager和TaskManager
- 2、并发度与Slots
- 3、开发环境搭建
- 4、提交到集群执行
- 5、并行度分析
- 6、Flink整体运行流程
Flink流式计算实战专题二
==楼兰
三、Flink运行架构
这一章重点是分析清楚运行架构以及并行度与slot的分配
1、JobManager和TaskManager
从之前的环境搭建过程中，也能够看到， Flink中的节点可以分为JobManager和TaskManager。
JobManager处理器也称为Master，用于协调分布式任务执行。他们用来调度task进行具体的任务。TaskManager处理器也称为Worker，用于实际执行任务。
一个有效的Flink集群中可以包含多个JobManager组成高可用集群，也可以有多个TaskManager进行并行计算。他们可以直接在物理机上启动，也可以通过像Yarn这样的资源调度框架启动。
每一个处理器都是一个单独的JVM进程，也可以通过配置的方式管理他们占用的内存资源。在flink-conf.yaml配置文件中，可以通过jobmanager.memory.process.size属性配置jobmanager占用的内存大小，taskmanager.memory.process.size属性配置每个taskmanager占用的内存大小。这个内存大小包含了JVM占用的堆内存以及堆外的元数据区和堆外直接内存的大小。这些参数也可以在提交任务的时候进行干预。
在一个典型的yarn集群中，JobManager在接收到任务时，整体执行的流程会是这样。
客户端会往JobManager提交任务，JobManager会往ResouceManager申请资源，当资源足够时，再将任务分配给集群中的TaskManager去执行。
只不过在Standalone模式下，这个ResourceManager是由Flink自己担任的。而在Yarn模式下，则是转为由Yarn来担任ResourceManager角色。
jobManager的内存架构：
Off-Heap Memory是Flink自行管理的一块内存，在JVM内存之外的直接内存Direct或者本地内存Native。配置JobManager的内存主要是两个参数 jobmanager.memory.flink.size 配置flink执行应用的内存大小 和jobmanager.memory.process.size flink集群运行以及应用执行的总内存大小。
更多详细配置参见官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/memory/mem_setup_jobmanager.html
Taskmanager的内存架构：
通常情况下，对于Taskmanager只需要配置一个总的内存大小即可。 taskmanager.memory.flink.size配置flink执行应用程序的总内存大小， taskmanager.memory.process.size 配置整个flink集群占用的内存大小。这两个属性不建议同时配置。
Flink也支持对taskManager的内存分布进行深度定制。TaskManager的Off-Heap Memory分为Managed Memory和Direct Memory两部分。 Managed Memory是Flink的TaskManager自行管理的一块内存，主要有三个用途：
1、Streaming Job可以用来缓存状态后端。 比如RocksDB 状态后端
2、 Batch Jobs主要用来进行排序、分组等结果的中间缓存。
3、User Defined Function可以直接用。 官网上说是给Python的processor用的。
Direct Memory部分分为Framework off-Heap 、Task off-Heap 、 Network三个部分。其中
1、Framework Heap Memory : Flink框架执行所需要的JVM堆内存，可以由属性 taskmanager.memory.framework.heap.size 配置。
2、Task off-Heap: 执行Flink程序所需要占用的JVM堆内存，可以由属性taskmanager.memory.task.heap.size 配置。
3、Network: 这一部分内存主要是为任务执行过程中的网络数据交互预留。这一部分是动态分配的，可以由以下三个属性进行定制： taskmanager.memory.network.min、taskmanager.memory.network.max、taskmanager.memory.network.fraction
这里只列出了使用频率相对比较高的几个配置信息。
更多详细配置参见官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/deployment/memory/mem_setup_tm.html
2、并发度与Slots
每一个TaskManager是一个独立的JVM进程，他可以在独立的线程上执行一个或多个任务task。为了控制一个taskManager能接收多少个task，TaskManager上就会划分出多个slot来进行控制。 每个slot表示的是TaskManager上拥有资源的一个固定大小的子集。flink-conf.yaml配置文件中的taskmanager.numberOfTaskSlots属性就配置了配个taskManager上有多少个slot。默认值是1，所以我们之前搭建的集群，有3个taskManager，集群内总共就只有3个slot。这些slot之间的内存管理也就是数据是相互隔离的。而这些slot其实都是在同一个JVM进程中，所以这里的隔离并不涉及到CPU等其他资源的隔离。
Task Slot是一个静态的概念，代表的是TaskManager具有的并发执行能力。另外还有一个概念并行度 parallelism就是一个动态的概念，表示的是运行程序时实际需要使用的并发能力。这个是可以在flink程序中进行控制的。如果集群提供的slot资源不够，那程序就无法正常执行下去，会表现为任务阻塞或者超时异常。
程序运行时的parallelism管理有三个地方可以配置，优先级最低的是在flink-conf.yaml文件中的parallelism.default这个属性，默认值是1。优先级较高的是在提交任务时可以指定任务整体的并行度要求。这个并行度可以在提交任务的管理页面和命令行中添加。 优先级最高的是在程序中指定的并行度。在flink的应用程序中，几乎每一个分布式操作都可以定制单独的并行度。这到底是是怎么回事呢？那现在我们就开发一个简单的flink应用了解一下。
3、开发环境搭建
flink提供了java和scala两套客户端API，我们这里采用java进行演示。
首先创建一个maven工程，在pom.xml文件中，引入客户端的依赖
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.12.5</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.12</artifactId><version>1.12.5</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>1.12.5</version></dependency>
后面这个依赖中最后的2.12表示是对应的scala版本。
然后就可以开发一个简单的flink应用程序。
package com.roy.flink.streaming;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class SocketWordCount {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();environment.setParallelism(1);final ParameterTool parameterTool = ParameterTool.fromArgs(args);String host = parameterTool.get("host");final int port = parameterTool.getInt("port");final DataStreamSource<String> inputDataStream = environment.socketTextStream(host, port);final DataStream<Tuple2<String, Integer>> wordcounts = inputDataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {final String[] words = value.split(" ");for (String word : words) {out.collect(new Tuple2<String, Integer>(word, 1));}}}).setParallelism(2).keyBy(value -> value.f0).sum(1).setParallelism(3);wordcounts.print();environment.execute("stream word count");}
}
这个程序的作用就是连接一个socket服务端，读取socketStream文本流，然后进行最为经典的WordCount操作。
首先，执行这个测试程序需要有一个socket服务端。 我们可以在hadoop01机器上使用nc指令模拟一个。 nc -lk 7777 在本地7777端口建立一个socket服务端。
然后，在本地的IDEA运行配置页面，指令要连接的host和port。
配置完成后，就可以在本地直接运行这个示例了。这也就是Flink所谓的LOCAL模式。
这个执行结果就是最终的wordcount结果。 但是，这里面有个有趣的现象。对hello的次数统计，是从1，一步一步统计到3的，而不是一次性统计到3。其实这也体现了流式计算的特点。这些词其实是一个一个统计的。
然后要注意一下我们代码中进行了多次setParallelism操作。在这个演示过程中，暂时没有体现出什么作用。在后续的演示中会有用。
4、提交到集群执行
这种本地执行的方式显然不具备生产使用的要求。我们可以使用maven进行编译，将这个代码编译成一个jar包，FlinkDemo-1.0.jar。
然后使用之前搭建的standalone模式的管理控制台来提交任务。
选standalone模式是因为这个模式的资源控制比较简单。
访问控制台，打开 Submit New Job页面，选择 +Add New 按钮，提交jar包。
单独提供一个jar包还并不足以启动任务，因为启动任务还需要指定任务的入口。选择这个FlinkDemo-1.0.jar，继续配置一个任务。
在这里注意下，提交任务时可以指定这个应用整体的Parallism 并行度。
点击提交，就可以开启一个任务。在running job页面就可以看到正在执行的任务 stream word count。选择这个任务，就能看到任务的执行情况。
这个数据流图展示了整个这个应用的具体执行的步骤。这些步骤整体就构成了数据流图。下面的数据流量会统计每个步骤经过的数据流量。在hadoop01机器上的nc服务中敲入字符，这个数据流量与记录数就会不断增加。
最后应用中通过print打印出来的消息会输入到应用的标准输出控制台。控制台的内容可以在TaskManagers菜单中查看。
5、并行度分析
这里我们重点分析每个蓝色方块下面的Parallelism参数。这里列出了每个步骤所占用的slot数量。而这里统计出来的slot数量就是按照之前所说的优先级确定的。整体优先级是这样。
程序中指定 > 提交任务时指定 > flink-conf.yaml中指定
注意下，这里面第一个和第四个步骤，他们的并行度其实是在具体实现时固定的指定了为1。
然后，我们回到Overview页面，查看下整体的slot情况。
接下来可以看到，我们这个job总共需要7个slot，但是集群中只有3个slot，程序也正常执行起来了。这也体现了slot复用的效果。也就是说slot可以在不同的执行步骤中处理不同的任务。只要集群资源能够支撑应用最大的并行度要求，整个应用就可以运行起来。实际上，Flink对于这个数据流图还会有一些自己的优化，例如某些相邻的操作，他们的并行度相同，任务也不是很复杂时，flink会将这些相邻的步骤进行合并。
这些slot在同一个任务内部是可以不断复用的，但是在不同的任务之间，是不能共用的。所以，这时可以看到，集群中仅有的3个slot已经全部被这个stream word count应用给占满了，如果需要再启动应用，就无法执行了。这时jobmanager会不断的尝试重新申请slot，如果集群中有空出来的slot，那就可以分配给应用。如果一直申请不下来，jobmanager会不断重试，默认每重试10次就会休息一点时间，过后再继续申请。如果在attached模式下，在客户端可以很清晰的看到这个过程。
6、Flink整体运行流程
然后我们再回头来看Flink官方提供的集群结构图就比较清晰了。
客户端
对于Flink，可以通过执行一个Java/Scala程序，或者通过./bin/flink run … 指令启动一个客户端。客户端将把sataflow提交给JobManager。客户端的主要作用其实就是构建好一个Dataflow graph或者也称为JobGraph，然后提交给客户端。而这个JobGraph如果在客户端本地构建，这就是Per-job模式，如果是提交到JobManager由Flink集群来构建，这就是Application模式。然后将提交完成后，客户端可以选择立即结束，这就是detached模式。也可以选择继续执行，来不断跟踪JobManager反馈的任务执行情况，这就是默认的attached模式。
JobManager
JobManager会首先接收到客户端提交的应用程序。这个应用程序整体会包含几个部分：作业图JobGraph，数据流图logic dataflow graph以及打包了所有类库以及资源的jar包。这些资源都将分发给所有的TaskManager去真正执行任务。
JobGraph相当于是一个设计图，之前Yarn的Per-job模式，往集群提交的就是这个JobGraph。JobManger会把JobGraph转换成一个物理层面的数据流图，这个图被叫做执行图 ExecutionGraph，这其中包含了所有可以并发执行的任务，相当于是一个执行计划。接下来JobGraph会向资源管理器 例如Yarn的ResourceManager 请求执行任务必要的资源，这些资源会表现为TaskManager上的slot插槽。一旦获得了足够多的资源，就会将执行图分发到真正运行任务的TaskManager上。而在运行过程中，JobManager还会负责所有需要中央协调的操作，例如反馈任务执行结果，协调检查点备份，协调故障恢复等。
JobManager整体上由三个功能模块组成：
ResourceManager
ResourceManager在Flink集群中负责申请、提供和注销集群资源，并且管理task slots。Flink中提供了非常多的ResourceManager实现，比如Yarn，Mesos，K8s和standalone模式。在standalone模式下，ResourceManager只负责在TaskManager之间协调slot的分配，而TaskManager的启动只能由TaskManager自己管理。
Dispatcher
Dispatcher模块提供了一系列的REST接口来提交任务，Flink的控制台也是由这个模块来提供。并且对于每一个执行的任务，Dispatcher会启动一个新的JobMaster，来对任务进行协调。
JobMaster
一个JobMaster负责管理一个单独的JobGraph。Flink集群中，同一时间可以运行多个任务，每个任务都由一个对应的JobMaster来管理。
一个集群中最少有一个JobManager。而在高可用部署时，也可以有多个JobManager。这些JobManager会选举出一个作为Leader。而其他的节点就出于StandBy备用的状态。
TaskManager
TaskManager也成为Worker。每个TaskManager上可以有一个或多个Slot。这些Slot就是程序运行的最小单元。 在flink.conf.yaml文件中通过taskmanager.numberOfTaskSlots属性进行配置。
每一个TaskManager就是一个独立的JVM进程，而每个Slot就会以这个进程中的一个线程执行。这些Slot在同一个任务中是共享的，一个Slot就足以贯穿应用的整个处理流程。Flink集群只需要关注一个任务内的最大并行数，提供足够的slot即可，而不用关注整个任务需要多少Slot。
这一个章节就整体讨论了Flink的运行机制，后面的章节我们就专注于学习Flink的客户端API了。
Flink流式计算从入门到实战 二相关推荐
- Flink流式计算从入门到实战 三
文章目录 四.Flink DataStream API 1.Flink程序的基础运行模型 2.Environment 运行环境 3.Source 3.1 基于File的数据源 3.2 基于Socket ...
- Flink流式计算从入门到实战 一
文章目录 一.理解Flink与流计算 1.初识Flink 2.Flink的适用场景 3.流式计算梳理 二.Flink安装部署 1.Flink的部署方式 2.获取Flink 3.实验环境与前置软件 4. ...
- Flink流式计算从入门到实战 四
文章目录 六.Flink Table API 和Flink SQL 1.Table API和SQL是什么? 2.如何使用Table API 3.基础编程框架 3.1 创建TableEnvironmen ...
- Flink 流式计算在节省资源方面的简单分析
本文由小米的王加胜同学分享,文章介绍了 Apache Flink 在小米的发展,从 Spark Streaming 迁移到 Flink ,在调度计算与调度数据.Mini batch 与 streami ...
- Flink系列之Flink流式计算引擎基础理论
声明: 文章中代码及相关语句为自己根据相应理解编写,文章中出现的相关图片为自己实践中的截图和相关技术对应的图片,若有相关异议,请联系删除.感谢.转载请注明出处,感谢. By luoye ...
- java8/Stream流式计算从入门到精通/函数式编程实战
摘要:Stream流式计算,本文讲解了Stream流式计算的概念,具体的使用步骤以及源码实现,最后讲解了使用Stream过程中需要注意的事项.Stream在公司项目中被频繁使用,在性能优化上具有广泛的 ...
- 大数据之Flink流式计算引擎
Flink DataFlow 数据的分类 有界数据 离线的计算 效率:硬件相同的情况下:时间 无界数据 实时的计算 效率:硬件环境相同的情况下,吞吐量:数据处理的数量.延迟:结果输出的时间-数据接收的 ...
- JStorm—实时流式计算框架入门介绍
JStorm介绍 JStorm是参考storm基于Java语言重写的实时流式计算系统框架,做了很多改进.如解决了之前的Storm nimbus节点的单点问题. JStorm类似于Hadoop ...
- Flink流式计算框架中的窗口函数
一.窗口(window) (1)一般真实的流都是无界的,怎样处理无界的数据? (2)可以把无限的数据流进行切分,得到有限的数据集进行处理--也就是得到有界流: (3)窗口(window)就是将无限流切 ...
最新文章
- 二.无显示器远程连接树莓派
- c++11- Alias Template
- 【Android】Looper消息分发(msg.target.dispatchMessage), Handler消息处理(消息回调/外部回调/自身回调)
- Bitbucket Cloud的新IP地址
- Nginx 配置内网访问树莓派4 ASP.NET Core 3.0 网站
- [机器学习-回归算法]一元线性回归用最小二乘法的推导过程
- iPhone 12 Pro拆解：韩国零部件占比最高 达26.8%
- 用SetCapture捕获鼠标消息
- [浪风分享]推荐一些不错的计算机书籍
- （转载）计算机英语名词简释
- MD文件阅读工具及配置
- 密码学：身份认证详解
- 软件开发常见的软件系统总结
- Kubernetes — 监控方案
- atomikos 配置好后 @transactional 注解不生效的问题
- 红帽RHCE考试下午-RHCE （RH294）任务概览[2021最新版]
- 51、【网工必备】两种常见网络拓扑图介绍，一看就会
- 个人博客标签和文章的表结构设计
- weblogic-cve_2020_2883漏洞复现
- 计算机文化基础清考时间,关于2012-2013学年第二学期开学初补考及各年级清考工作安排的通知...