【赠Flink书籍】Flink on YARN/K8S 部署实战教程(书摘)

上次推荐机械工业出版社华章公司出版的Flink新书《Flink原理、实战与性能优化》，反响很不错。这次华章公司为感谢大家对《Flink原理、实战与性能优化》的热情与关注，特地拿出 两本 书来赠与Flink公众号的订阅者。

规则：说出为什么你想要这本书，点赞数排名前 两名 公众号后台私我你的联系方式。截止时间：5月25日 06:00 。

以下是《Flink原理、实战与性能优化》的部分书摘

Flink on K8S 部署

容器化部署是目前业界非常流行的一项技术，基于 Docker 镜像部署，能够更加方便对应用进行管理和运维。同时随着Docker容器编排工具Kubernetes近几年逐渐火热起来，目前大多数企业通过Kubernetes来管理容器资源。而Flink也在最新的版本中支持了在Kubernetes部署Flink集群，可以让用户基于Kubunetes来构建Flink Session Cluster，也可以直接通过Docker镜像的方式向Kuernetes集群中提交独立的Flink任务。以下我们将介绍如何基于Kubunetes部署Flink集群，如果用户已经有现成的Kubernetes集群则可以直接使用，否则搭建Kubernetes集群，具体的搭建步骤可以参考Kubernetes官方网站，这里我们将不再进行深入介绍。

1.Yaml配置

在使用Kubernetes上构建Flink Session Cluster的过程中，需要将Flink中包含的组件分别在Kubernetes集群中启动对应的Docker镜像，其中包括JobManager，TaskManager，JobManagerServices三个镜像服务，其中每个镜像服务都可以从中央镜像仓库中获取，也可构建本地的镜像仓库。

JobManager yaml配置

主要提供运行JobManager组件的参数配置，以及JobManager自身参数，例如RPC端口等配置信息。

TaskManager Yaml配置

主要提供运行JobManager组件的参数配置，以及JobManager自身参数，例如RPC端口等配置信息。

JobManagerServices配置

主要目的是为了能够提供对外的RestApi和UI地址，通过Flink UI的方式访问集群并获取任务以及监控信息。 代码清单1-1 JobManagerServices配置文件

启动Flink Sesssion Cluster

当服务配置文件编辑完毕后，就可以通过使用Kubectl命令创建Flink Session Cluster，启动完成后通过JobJobManagerServices中配置的WebUI端口，访问FlinkWeb页面。

获取Flink Session Cluster状态

可以通过kubectl proxy按以下方式访问Flink UI，需要保证kubectl proxy在终端中运行，然后在浏览器里输入以下地址，就能够访问FlinkUI。

停止Flink Session Cluster

可以通过kubectl delete命令行来停止Flink Session Cluster。

Flink on YARN 部署

Hadoop Yarn是 Hadoop 2.x提出的统一资源管理器，也是目前在相对比较流行的大数据资源管理平台，类似Spark和MapReduce等分布式处理框架都会兼容运行在Yarn上，Flink也是如此。 Flink应用提交到Yarn上目前有两种模式，一种是Yarn Session Model，Flink会向Hadoop Yarn上申请足够多的资源，然后在Yarn启动长时间运行的Flink集群，用户可以通过RestAPI或者Web页面将Flink任务提交到Flink Session集群上运行。Single Job Model和有多数框架使用模式类似，每个Flink任务单独向Yarn提交一个Application，然后每个任务都有自己的JobManager和TaskManager，任务结束后对应的组件都会跟着任务释放。

图1-1 Flink On Yarn运行流程图

环境依赖

将Flink部署在Yarn之前，需要对环境有以下两点配置要求，用户需要确认Hadoop环境是否满足要求，主要有以下两点需要确认：

• Hadoop版本至少保证在2.2以上，集群中安装有HDFS服务。

• 主机中配置HADOOPCONFDIR变量，并在对应的路径中有Hadoop配置文件，主要包括hdfs-default.xml、hdfs-site.xml以及yarn-site.xml等配置文件。在启动Flink集群的过程中，Flink会通过识别HADOOPCONFDIR环境变量，读取Hadoop配置参数。

集群安装

用户从Flink官方下载地址下载Flink安装包，选择一台具有Hadoop客户端配置的主机，通过解压操作并进入到安装路径中，这样基本就完成对Flink集群的安装。

Yarn Session模式

Yarn Session 模式其实是在Yarn上启动一个Flink集群，其中包括JobManager和TaskManager组件。Session集群会一直运行在Hadoop Yarn之上，底层对应的是Hadoop的一个Yarn Application应用，Yarn Session Cluster启动后，用户就能够通过命令行或RestAPI等方式向Yarn Session集群中提交Flink任务，不需要再和Yarn进行交互。

启动Yarn Session

启动Yarn Session Cluster之前需要使用Hadoop客户端参数，Flink默认是用YARNCONFDIR或者HADOOPCONFDIR环境变量中进行获取Hadoop客户端配置文件，如果启动的节点中没有相应的环境变量和配置文件则可能导致Flink启动过程无法正常连接到Hadoop Yarn集群。 如果节点中没有对应的环境变量，则建议用户也可以在每次启动Yarn Session的之前通过手动的方式对环境变量进行赋值即可。启动节点中如果没有Hadoop客户端配置，则建议用户实现将配置从Hadoop集群中获取出来，然后放置在当前指定路径中，然后通过上述步骤进行配置环境变量配置，然后启动Flink Yarn Session集群。 通过如下yarn-session.sh命令启动Flink Yarn Session 集群，其中-n参数表示启动4个Yarn Container，-jm参数配置JobManager的Jvm内存大小，-tm参数配置TaskManager的内存大小，-s表示集群中共启动16个slots来处理Task任务。

启动完毕之后就可以通过在Yarn的任务管理页面查看Flink Session集群状况，并点击ApplicationMaster对应URL，进入到Flink Session Cluster集群中。

Yarn Session独立模式

通过上述方式启动Yarn Session集群，集群的运行与管理依赖于于本地Yarn Session集群的启动进程，一旦进程关闭，则整个Session集群也会终止。此时可以通过参数进行配置可以通过在启动Session过程中指定参数--d或者--detached，使得启动的Session集群交给Yarn集群管理，与本地提交进程脱离。通过这种模式下启动Flink集群时，当停止Flink Session Cluster时，需要通过Yarn Application -kill [appid]来终止Flink Session集群。

Attach to an existing Session

和Detached Yarn Session相反，如果用户想绑定到Yarn上已经提交的Session，这个时候就需要Attach本地的进程关联到Yarn集群对应的Application。通过以上命令Yarn集群上ApplicationID对应的的Session就能够绑定到本地进程中，这样用户就能够对Session进行操作，包括执行停止命令等，执行停止命令可以通过Ctrl+C命令，或者输入stop命令，就会将Flink Session Cluster停止。

Submit job to Session

当Flink Yarn Session集群构建好之后，就可以向Session集群中提交Flink任务。可以通过命令行的方式提交Flink应用到Session集群中。通过以下命令将Flink任务就能够提交到Session中，正常情况下，就能够直接进入到Flink监控页面查看到已经提交的Flink任务。 ./bin/flink run ./windowsWordCountApp.jar

容错配置

在Yarn上执行的Flink Session通常情况下需要进行对应的任务恢复策略配置，以防止因为系统某些系统问题导致整个集群出现异常。针对在Yarn上的部署，Flink单独提供了几个相关的参数如下，用户可以根据实际情况进行配置：

yarn.reallocate-failed：

该参数表示集群中TaskManager失败是否被重新拉起，设定为True则表示重新分配资源并拉起失败的TaskManager，默认为True，底层其实是Yarn是否重新分配TaskManager的Container。

• yarn.maximum-failed-containers： 该参数表示集群所容忍最大失败的Container数量，如果超过设定参数，则会直接导致整个Session集群失败并停止。参数默认值为TaskManager数量，也就是提交任务时-n参数对应的值。

• yarn.application-attempts： 该参数表示整个Session集群对应的Yarn Application失败重启的次数，如果Session集群所在的整个应用失败，则在该参数范围内，Yarn也会重新拉起对应Session的Application。如果重启次数超过该参数，Yarn不会再重启应用，整个Flink Session则失败，此时Session上提交的任务也全会全部失败。 对于以上三个参数，实际上是从不同层面地保证Flink任务在Yarn集群上正常运行，这些参数都可以通过在conf/flink-default.yaml文件中进行配置，也可以在启动Session集群时通过-D动态参数指定，如-Dyarn.application-attempts=10。

防火墙配置

针对有些用户的集群在生产环境中，具有非常高的安全管控，针对网络安全基本都是通过防火墙进行隔离，但是有些情况下如果用户想要实现集群之外的机器上远程提交Flink作业，在Standalone集群中比较容易实现，因为Rpc端口和Rest端口都能够通过防火墙配置即可，而对于Yarn Session Cluster每次启动一次Session，Yarn都会给相应的Application分配一个随机的端口，这使得Flink Session中JobManager的Rest和Rpc端口都会发生变换，而客户端无法自动感知远程Session Cluster端口的变换，而可能被防火墙隔离掉，无法连接到Session Cluster集群，进而不能提交任务到集群。针对这种情况Flink提供了相应解决策略，就是通过端口段实现。

日志配置

用户通过Yarn Session构建Flink集群的模式，一定程度上减少了用户在使用Flink Cluster的开销，但是不可避免也会带来一些额外的问题，例如比较常见的表示提交的应用该如何进行监控和问题排查。应用提交Session后执行在Yarn的过程基本上都被屏蔽掉，用户无法有效获取相关的日志信息。Session部署失败可能有很多原因，主要原因可能有Hadoop配置问题，例如HDFS权限问题或者Yarn参数配置问题，都有可能导致Session失败，另外一方面因为用户编译的Flink版本和集群Yarn的版本不一致也会导致Session部署失败，需要用户从Flink官方获得相应版本Flink安装包，防止出现因为版本出现的问题。与此同时Flink也可能因为其他的原因导致失败，在这里提供几个相对比较高效定位问题的方式，方便用户解决问题。

Session日志查看

由于Flink Session是提交在Hadoop上面的，Session也是Hadoop Yarn的一个任务，因此对于任务日志的收集其实和Hadoop Yarn提供的方式是一致的，也就是通过Yarn提供的日志聚合功能完成对Session的日志收集。当Session Cluster关闭后，可以通过如下命令行对对日志进行回收，然后用户通过编译器打开查看：

Job日志查看

用户通过Flink客户端命令行或者RestApi提交的任务，在Flink Session集群中对应的就是不同的Job，Job和Job之间资源相互隔离，另外Job都是通过Task单元处理具体的数据，而Task是共享在Flink Session集群的资源。因此需要注意的是每个Job的Task日志分别在不同的TaskManager执行，目前Flink的任务只能讲日志区分在TaskManager层面，因此在相同的TaskManager上执行的任务的日志会记录在一起，相对查看起来不是特别方面，需要用户自己筛选。

Single Job模式

在Single Job模式中，Flink任务可以直接以单个应用提交到Yarn上，而无需使用Session模式提交，每次提交的Flink任务就是一个独立的Yarn Application，且在每个任务重都会有自己的JobManager和TaskManager组件，所有的资源独立使用。 可以通过以下命令直接将任务提交到Hadoop Yarn集群，建立对应的Flink Application。需要在参数中指定-m yarn-cluster，表示使用Yarn集群提交Flink任务，-yn表示需要的TaskManager数量。