flink学习（scala版)--第三章：Flink的部署模式_单作业模式

1.flink作业的部署方式：        

        上节介绍到通过本地、服务器单节点启动以及集群启动的方式运行程序，本节将介绍一下flink作业的三种部署模式

       在一些应用场景中，对于集群资源分配和占用的方式，可能会有特定的需求。Flink为各种场景提供了不同的任务部署模式，主要有以下三种：

        ⚫会话模式（Session Mode）

        ⚫单作业模式（Per-Job Mode）

        ⚫应用模式（Application Mode）

        它们的区别主要在于：集群的生命周期以及资源的分配方式；以及应用的main方法到底在哪里执行——客户端（Client）还是JobManager。 

        1.1 会话模式（Session Mode）

        会话模式其实最符合常规思维。我们需要先启动一个集群，保持一个会话，在这个会话中通过客户端提交作业，如图所示。集群启动时所有资源就都已经确定，所以所有提交的作业会竞争集群中的资源。

会话模式比较适合于单个规模小、执行时间短的大量作业。

        这样的好处很明显，我们只需要一个集群，就像一个大箱子，所有的作业提交之后都塞进去；集群的生命周期是超越于作业之上的，铁打的营盘流水的兵，作业结束了就释放资源，集群依然正常运行。当然缺点也是显而易见的：因为资源是共享的，所以资源不够了，提交新的作业就会失败。另外，同一个TaskManager上可能运行了很多作业，如果其中一个发生故障导致TaskManager宕机，那么所有作业都会受到影响。

1.2 单作业模式（Per-Job Mode）

         会话模式因为资源共享会导致很多问题，所以为了更好地隔离资源，我们可以考虑为每个提交的作业启动一个集群，这就是单作业（Per-Job）模式，如图

        单作业模式也很好理解，就是严格的一对一，集群只为这个作业而生。同样由客户端运行应用程序，然后启动集群，作业被提交给JobManager，进而分发给TaskManager执行。作业完成后，集群就会关闭，所有资源也会释放。这样一来，每个作业都有它自己的JobManager管理，占用独享的资源，即使发生故障，它的TaskManager宕机也不会影响其他作业。

        这些特性使得单作业模式在生产环境运行更加稳定，所以是实际应用的首选模式。

         需要注意的是，Flink本身无法直接这样运行，所以单作业模式一般需要借助一些资源管理平台来启动集群，比如YARN、Kubernetes。

 1.3 应用模式（Application Mode）

        前面提到的两种模式下，应用代码都是在客户端上执行，然后由客户端提交给JobManager的。但是这种方式客户端需要占用大量网络带宽，去下载依赖和把二进制数据发送给JobManager；加上很多情况下我们提交作业用的是同一个客户端，就会加重客户端所在节点的资源消耗。

        所以解决办法就是，不要客户端了，直接把应用提交到JobManger上运行。而这也就代表着，我们需要为每一个提交的应用单独启动一个JobManager，也就是创建一个集群。这个JobManager只为执行这一个应用而存在，执行结束之后JobManager也就关闭了，这就是应用模式，如图所示。

         应用模式与单作业模式，都是提交作业之后才创建集群；单作业模式是通过客户端来提交的，客户端解析出的每一个作业对应一个集群；而应用模式下，是直接由JobManager执行应用程序的，并且即使应用包含了多个作业，也只创建一个集群。

2.flink服务的部署方式：

2.1独立模式（Standalone）

        独立模式（Standalone）是部署Flink最基本也是最简单的方式：所需要的所有Flink组件，都只是操作系统上运行的一个JVM进程。

        独立模式是独立运行的，不依赖任何外部的资源管理平台；当然独立也是有代价的：如果资源不足，或者出现故障，没有自动扩展或重分配资源的保证，必须手动处理。所以独立模式一般只用在开发测试或作业非常少的场景下。

2.1.1会话模式部署

        独立模式的特点是不依赖外部资源管理平台，而会话模式的特点是先启动集群、后提交作业。所以，我们在上节用的就是独立模式（Standalone）的会话模式部署。

2.1.2单作业模式部署

        Flink本身无法直接以单作业方式启动集群，一般需要借助资源管理平台。所以Flink的独立（Standalone）集群并不支持单作业模式部署。

2.1.3应用模式部署

        应用模式下不会提前创建集群，所以不能调用start-cluster.sh脚本。我们可以使用同样在bin目录下的standalone-job.sh来创建一个JobManager。

        具体步骤如下：（1）进入到Flink的安装路径下，将应用程序的jar包放到lib/目录下。

$cp ./flinkTutorial-1.0-SNAPSHOT.jar lib/

        （2）执行以下命令，启动JobManager。

$ ./bin/standalone-job.shstart --job-classname com.flink.test.StreamWordCount

        这里我们直接指定作业入口类，脚本会到lib目录扫描所有的jar包。

        （3）同样是使用bin目录下的脚本，启动TaskManager。

$ ./bin/taskmanager.shstart

       （4）如果希望停掉集群，同样可以使用脚本，命令如下。

$ ./bin/standalone-job.shstop
$ ./bin/taskmanager.shstop

2.2YARN模式

        整体来说，YARN上部署的过程是：客户端把Flink应用提交给Yarn的ResourceManager, Yarn的ResourceManager会向Yarn的NodeManager申请容器。在这些容器上，Flink会部署JobManager和TaskManager的实例，从而启动集群。Flink会根据运行在JobManger上的作业所需要的slots数量动态分配TaskManager资源。

2.2.1 相关准备和配置

        在Flink1.8.0之前的版本，想要以YARN模式部署Flink任务时，需要Flink是有Hadoop支持的。从Flink1.8版本开始，不再提供基于Hadoop编译的安装包，若需要Hadoop的环境支持，需要自行在官网下载Hadoop相关版本的组件flink-shaded-hadoop-2-uber-2.7.5-10.0.jar，并将该组件上传至Flink的lib目录下。在Flink 1.11.0版本之后，增加了很多重要新特性，其中就包括增加了对Hadoop3.0.0以及更高版本Hadoop的支持，不再提供“flink-shaded-hadoop-*”jar包，而是通过配置环境变量完成与YARN集群的对接。

        在将Flink任务部署至YARN集群之前，需要确认集群是否安装有Hadoop，保证Hadoop版本至少在2.2以上，并且集群中安装有HDFS服务。

        具体配置步骤如下：

        （1）解压flink安装包，并将解压后的安装包重命名为flink-1.13.0-yarn，本节的相关操作都将默认在此安装路径下执行。（flink解压参考上节）

        （2）配置环境变量，增加环境变量配置如下,其他节点也要同步如下配置：

$  sudo vim /etc/profile.d/my_env.sh

export HADOOP_HOME=/opt/module/hadoop-3.1.3
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
export HADOOP_CLASSPATH=`hadoop classpath`

        （3）启动Hadoop集群，包括HDFS和YARN。

                分别在3台节点服务器查看进程启动情况。

        （4）进入conf目录，修改flink-conf.yaml文件，修改以下配置，这些配置项的含义在进行Standalone模式配置的时候进行过讲解，若在提交命令中不特定指明，这些配置将作为默认配置。

$ cd /opt/module/flink-1.13.0-yarn/conf/
$ vim flink-conf.yaml
jobmanager.memory.process.size: 1600m
taskmanager.memory.process.size: 1728m
taskmanager.numberOfTaskSlots: 8
parallelism.default: 1

2.2.2 会话模式部署

        YARN的会话模式与独立集群略有不同，需要首先申请一个YARN会话（YARN session）来启动Flink集群。具体步骤如下：

        1. 启动集群

        （1）启动Hadoop集群，包括HDFS和YARN。

        （2）执行脚本命令向YARN集群申请资源，开启一个YARN会话，启动Flink集群。

$ bin/yarn-session.sh -nm test

可用参数解读：

        ⚫-d：分离模式，如果你不想让FlinkYARN客户端一直前台运行，可以使用这个参数，即使关掉当前对话窗口，YARN session也可以后台运行。

        ⚫-jm(--jobManagerMemory)：配置JobManager所需内存，默认单位MB。

        ⚫-nm(--name)：配置在YARNUI界面上显示的任务名。

        ⚫-qu(--queue)：指定YARN队列名。

        ⚫-tm(--taskManager)：配置每个TaskManager所使用内存。

        注意：Flink1.11.0版本不再使用-n参数和-s参数分别指定TaskManager数量和slot数量，YARN会按照需求动态分配TaskManager和slot。所以从这个意义上讲，YARN的会话模式也不会把集群资源固定，同样是动态分配的。

        YARN Session启动之后会给出一个webUI地址以及一个YARN applicationID，如下所示，用户可以通过webUI或者命令行两种方式提交作业。

        

        2. 提交作业

        （1）通过WebUI提交作业

                这种方式比较简单，与上文所述Standalone部署模式基本同。        

        （2）通过命令行提交作业

                ①将Standalone模式讲解中打包好的任务运行JAR包上传至集群 

                ②执行以下命令将该任务提交到已经开启的Yarn-Session中运行。

$ ./bin/flink run -c com.flink.test.StreamWordCount ./jar/flinkToturial-1.0-SNAPSHOT.jar --host localhost --port 7777

        客户端可以自行确定JobManager的地址，也可以通过-m或者-jobmanager参数指定JobManager的地址，JobManager的地址在YARN Session的启动页面中可以找到。

                ③任务提交成功后，可在YARN的WebUI界面查看运行情况。

        如图所示，从图中可以看到我们创建的Yarn-Session实际上是一个Yarn的Application，并且有唯一的ApplicationID。 

        ④也可以通过Flink的Web UI页面查看提交任务的运行情况。

2.2.3 单作业模式部署

        在YARN环境中，由于有了外部平台做资源调度，所以我们也可以直接向YARN提交一个单独的作业，从而启动一个Flink集群。

        （1）执行命令提交作业。

$  bin/flink  run -d -t  yarn-per-job -c com.flink.test.StreamWordCount ./jar/flinkToturial-1.0-SNAPSHOT.jar

        （2） 同样可以在YARN的ResourceManager界面查看执行情况

                在任务提交成功后，控制台打印的日志会提供FlinkWebUI页面地址，我们可以打开Flink Web UI页面进行监控

        （3）可以使用命令行查看或取消作业，命令如下。

$ ./bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY
$ ./bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY <jobId>

        这里的application_XXXX_YY是当前应用的ID，<jobId>是作业的ID。注意如果取消作业，整个Flink集群也会停掉。

2.2.4 应用模式部署

       应用模式同样非常简单，与单作业模式类似，直接执行flinkrun-application命令即可。

        （1）执行命令提交作业。

$ bin/flink run-application-t yarn-application -c com.flink.test.StreamWordCount ./jar/flinkToturial-1.0-SNAPSHOT.jar

        （2）在命令行中查看或取消作业。

$./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
$./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>

       （3）也可以通过yarn.provided.lib.dirs配置选项指定位置，将jar上传到远程。

$ ./bin/flink run-application -t  yarn-application-Dyarn.provided.lib.dirs="hdfs://myhdfs/my-remote-flink-dist-dir"hdfs://myhdfs/jars/my-application.jar

        这种方式下jar可以预先上传到HDFS，而不需要单独发送到集群，这就使得作业提交更加轻量了。

2.3 K8S 模式

        容器化部署是如今业界流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（K8S），而Flink也在最近的版本中支持了K8S部署模式。基本原理与YARN是类似的，具体配置可以参见官网说明，目前在国内应用也不是很多。