Flink集群部署详细步骤

桃扇骨 2022-05-19 14:12 800阅读 0赞

### **1.下载Flink压缩包** ###

下载地址：[http://flink.apache.org/downloads.html][http_flink.apache.org_downloads.html]。

我集群环境是hadoop2.6，Scala2.11版本的，所以下载的是：

flink-1.3.1-bin-hadoop26-scala\_2.11.tgz。

### 2.**解压** ###

上传至五个节点的相同目录，执行如下命令解压：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;"> <p>tar xzf flink-1.3.1-bin-hadoop26-scala_2.11.tgz</p> </td> 
  </tr> 
 </tbody> 
</table>

### 3.**配置master节点** ###

选择一个 master节点(JobManager)然后在conf/flink-conf.yaml中设置jobmanager.rpc.address 配置项为该节点的IP 或者主机名。确保所有节点有有一样的jobmanager.rpc.address 配置。

### 4.**配置slaves** ###

将所有的 worker 节点 （TaskManager）的IP 或者主机名（一行一个）填入conf/slaves 文件中。

### 5.**启动flink集群** ###

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;"> <p>bin/start-cluster.sh</p> </td> 
  </tr> 
 </tbody> 
</table>

### 6.**注意** ###

问https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/config.html查看更多可用的配置项。为了使Flink 更高效的运行，还需要设置一些配置项。

以下都是非常重要的配置项：

1、TaskManager总共能使用的内存大小（taskmanager.heap.mb）

2、每一台机器上能使用的 CPU 个数（taskmanager.numberOfTaskSlots）

3、集群中的总 CPU个数（parallelism.default）

4、临时目录（taskmanager.tmp.dirs）

### **7.示例** ###

下面的例子阐述了三个节点的集群部署（IP地址从10.0.0.1 到10.0.0.3，主机名分别为master, worker1, worker2）。并且展示了配置文件，以及所有机器上一致的可访问的安装路径。

### 8.Flink On Yarn配置![Center][] ###

你可以很方便地将 Flink部署在现有的YARN集群上，操作如下：

1、下载 Flink Hadoop2包: Flink with Hadoop 2

2、确保你的 HADOOP\_HOME (或YARN\_CONF\_DIR 或HADOOP\_CONF\_DIR) \_\_环境变量\_\_设置成你的YARN 和HDFS 配置。

3、运行 YARN客户端：./bin/yarn-session.sh。你可以带参数运行客户端 -n 10 -tm 8192表示分配 10个 TaskManager，每个拥有8 GB 的内存。

# flink集群安装部署 #

## yarn集群模式 ##

*  快速开始

1.  在yarn上启动一个一直运行的flink集群
2.  在yarn上运行一个flink job

*  flink yarn session

1.  启动flink session
2.  提交任务到flink

*  在yarn上运行一个独立的flink job

1.  用户依赖jar包和classpath

*  flink on yarn的故障恢复
 *  调试一个失败的yarn session

1.  日志文件
2.  yarn client控制台和web界面

*  针对指定的hadoop版本构建yarn client
 *  在yarn上运行flink使用防火墙
 *  flink on yarn 内部实现

### 快速开始 ###

在yarn上启动一个一直运行的flink集群

启动一个yarn session使用4个taskmanager(每个节点4GB内存)  
注意：如果自己的虚拟机没有这么大内存的话，可以吧-n设置小一点，对应的后面的内存-jm -tm也设置小一点，否则，如果内存不够用，会导致启动失败。

1.  `tar xvzf flink-1.4.2-bin-hadoop2.tgz`
2.  `cd flink-1.4.2/`
3.  `./bin/yarn-session.sh -n 4 -jm 1024 -tm 4096`

通过-s参数指定每一个taskmanager分配多少个slots(处理进程)。我们建议设置为每个机器的CPU核数。

一旦session创建成功，你可以使用./bin/flink工具向集群提交任务。

在yarn上运行一个flink job

1.  `tar xvzf flink-1.4.2-bin-hadoop2.tgz`
2.  `cd flink-1.4.2/`
3.  `./bin/flink run -m yarn-cluster -yn 4 -yjm 1024 -ytm 4096 ./examples/batch/WordCount.jar`

### flink yarn session ###

yarn是一个集群资源管理框架。它运行在集群之上运行各种分布式应用程序。flink像其他程序一样，也可以在yarn上运行。用户不需要设置或者安装任何东西，如果已经有一个安装配置好的yarn。

必须的依赖

*  至少是hadoop2.2
 *  hdfs（或者是其它支持hadoop的分布式文件系统）

如果你在使用flink yarn client的时候有什么问题，[可以到这里查找答案][Link 1]。

启动flink session

按照下面的步骤来学习如何在yarn集群上启动一个flink session

一个session将会包含所有必须的flink 服务(jobmanager和taskmanager)，这样你就可以向这个集群提交程序了。注意：每个session会话你可以运行多个程序。

下载flink

下载hadoop2对应的flink安装包，[点此下载][http_flink.apache.org_downloads.html]。它包含了必须的文件。

使用下面命令解压：

1.  `tar xvzf flink-1.4.2-bin-hadoop2.tgz`
2.  `cd flink-1.4.2/`

启动一个session

使用下面命令启动一个session

./bin/yarn-session.sh

这个命令将会输出下面内容：

1.  `用法:`
2.  `必选`
3.  `-n,--container <arg> 分配多少个yarn容器 (=taskmanager的数量)`
4.  `可选`
5.  `-D <arg> 动态属性`
6.  `-d,--detached 独立运行`
7.  `-jm,--jobManagerMemory <arg> JobManager的内存 [in MB]`
8.  `-nm,--name 在YARN上为一个自定义的应用设置一个名字`
9.  `-q,--query 显示yarn中可用的资源 (内存, cpu核数)`
10. `-qu,--queue <arg> 指定YARN队列.`
11. `-s,--slots <arg> 每个TaskManager使用的slots数量`
12. `-tm,--taskManagerMemory <arg> 每个TaskManager的内存 [in MB]`
13. `-z,--zookeeperNamespace <arg> 针对HA模式在zookeeper上创建NameSpace`

**请注意：client必须要设置YARN\_CONF\_DIR或者HADOOP\_CONF\_DIR环境变量，通过这个环境变量来读取YARN和HDFS的配置信息，否则启动会失败。**

**经试验发现，其实如果配置的有HADOOP\_HOME环境变量的话也是可以的。HADOOP\_HOME ，YARN\_CONF\_DIR，HADOOP\_CONF\_DIR 只要配置的有任何一个即可。**

例子：下面的命令会申请10个taskmanager，每个8G内存和32个solt

./bin/yarn-session.sh -n 10 -tm 8192 -s 32

该系统默认会使用这个配置文件：conf/flink-conf.yaml。如果你想修改一些参数，请查看我们的[配置指南][Link 2]。

flink on yarn模式将会覆盖一些配置文件 jobmanager.rpc.address(因为jobmanager总是分配在不同的机器)，taskmanager.tmp.dirs(我们使用yarn提供的临时目录)和parallelism.default 如果solts的数量已经被指定。

如果你不想修改配置文件去改变参数，有一个选择是通过动态的参数-D 来指定。所以你可以传递参数：-Dfs.overwrite-files=true -Dtaskmanager.network.memory.min=536346624

上面的例子将会启动11个容器(即使仅请求10个容器)，因为有一个额外的容器来启动ApplicationMaster 和 job manager

一旦flink在你的yarn集群上部署，它将会显示job manager的连接详细信息。

停止yarn session通过停止unix进程(使用CTRL+C）或者在client中输入stop。

Flink on yarn只会启动请求的资源，如果集群资源充足。大多数yarn调度器请求容器的内存，一些也会请求cpu。默认，cpu的核数等于slots的数量，通过-s参数指定。这个参数yarn.containers.vcores的值允许使用一个自定义值来进行覆盖。

后台 yarn session

如果你不希望flink yarn client一直运行，也可以启动一个后台运行的yarn session。使用这个参数：-d 或者 --detached

在这种情况下，flink yarn client将会只提交任务到集群然后关闭自己。注意：在这种情况下，无法使用flink停止yarn session。

使用yarn 工具 来停止yarn session

yarn application -kill <applicationId>

附着到一个已存在的session

使用下面命令启动一个session

./bin/yarn-session.sh

执行这个命令将会显示下面内容：

1.  `用法:`
2.  `必须`
3.  `-id,--applicationId <yarnAppId> YARN集群上的任务id`

正如前面提到的，YARN\_CONF\_DIR或者HADOOP\_CONF\_DIR环境变量必须是可以读取到YARN和HDFS配置的。

例如：发出下面命令可以附着到一个运行中的flink yarn session

./bin/yarn-session.sh -id application_1463870264508_0029

附着到一个运行的session使用yarn resourcemanager来确定job Manager 的RPC端口。

停止yarn session通过停止unix进程(使用CTRL+C）或者在client中输入stop

提交任务到flink

使用下面的命令提交一个flink程序到yarn集群

./bin/flink

请参考[客户端命令行操作文档][Link 3]

这个命令将会向你展示一个这样一个帮助菜单

"run" 参数可以编译和运行一个程序

1.  `用法: run [OPTIONS] <jar-file> <arguments>`
2.  `"run" 操作参数:`
3.  `-c,--class <classname> 如果没有在jar包中指定入口类，则需要在这里通过这个参数指定`
4.   
5.  `-m,--jobmanager <host:port> 指定需要连接的jobmanager(主节点)地址`
6.  `使用这个参数可以指定一个不同于配置文件中的jobmanager`
7.   
8.  `-p,--parallelism <parallelism> 指定程序的并行度。可以覆盖配置文件中的默认值。`

使用run 命令向yarn集群提交一个job。客户端可以确定jobmanager的地址。当然，你也可以通过-m参数指定jobmanager。jobmanager的地址在yarn控制台上可以看到。

例子：【注意：下面的命令官网文档提供的有问题，执行失败】

1.  `wget -O LICENSE-2.0.txt http://www.apache.org/licenses/LICENSE-2.0.txt`
2.  `hadoop fs -copyFromLocal LICENSE-2.0.txt hdfs:/// ...`
3.  `【注意：下面的命令官网文档提供的有问题，执行失败】`
4.  `./bin/flink run ./examples/batch/WordCount.jar \`
5.  `        hdfs:///..../LICENSE-2.0.txt hdfs:///.../wordcount-result.txt`
6.  `查看flink源码发现，wordCount.jar可以不提供参数，或者提供参数，提供参数的时候需要使用input和output参数指定：`
7.  `上面的命令需要修改为如下格式才能正常执行[传递的两个参数需要使用-input和 -output来指定]`
8.  `./bin/flink run ./examples/batch/WordCount.jar -input hdfs://hostname:port/LICENSE-2.0.txt -output hdfs://hostname:port/wordcount-result.txt`
9.   
10.

如果有以下错误，确保所有taskmanager是否已经启动：

1.  `Exception in thread "main" org.apache.flink.compiler.CompilerException:`
2.  `Available instances could not be determined from job manager: Connection timed out.`

你可以在jobmanager的web界面上检查taskmanager的数量。这个web界面的地址会打印在yarn session的控制台上。

如果没有发现taskmanager，你应该通过日志文件来检查问题。

### 在yarn上运行一个独立的flink job ###

这个文档描述了如何在一个hadoop yarn环境中启动flink集群。也可以在yarn中启动只执行单个任务的flink。

请注意：client期望设置-yn 参数(taskmanager的数量)

例子：

./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar

yarn session命令行的选项也可以使用./bin/flink 工具获得。它们都有一个y或者yarn的前缀

注意：通过为每个任务设置不同的环境变量 FLINK\_CONF\_DIR，可以为每个任务使用不同的配置目录。从 Flink 分发包中复制 conf 目录，然后修改配置，例如，每个任务不同的日志设置

用户依赖jar包和classpath

默认情况下，当运行一个独立的job的时候，这个flink job将包含用户依赖的jar包。可以通过参数**yarn.per-job-cluster.include-user-jar**来控制。

当设置为 DISABLED ，flink将会包含用户classpath下面的jar包。

用户jar包在类路径中的位置可以通过下面参数来控制：

1.  `* ORDER: (默认) 添加jar包到系统类路径下面，按照字典顺序.`
2.  `* FIRST: 将jar添加到类路径的前面.`
3.  `* LAST: 将jar添加到类路径的最后.`

## ##

### flink on yarn的故障恢复 ###

flink 的 yarn 客户端通过下面的配置参数来控制容器的故障恢复。这些参数可以通过conf/flink-conf.yaml 或者在启动yarn session的时候通过-D参数来指定。

*  yarn.reallocate-failed：这个参数控制了flink是否应该重新分配失败的taskmanager容器。默认是true。
 *  yarn.maximum-failed-containers：applicationMaster可以接受的容器最大失败次数，达到这个参数，就会认为yarn session失败。默认这个次数和初始化请求的taskmanager数量相等(-n 参数指定的)。
 *  yarn.application-attempts：applicationMaster重试的次数。如果这个值被设置为1(默认就是1)，当application master失败的时候，yarn session也会失败。设置一个比较大的值的话，yarn会尝试重启applicationMaster。

### 调试一个失败的yarn session ###

一个flink yarn session部署失败可能会有很多原因。一个错误的hadoop配置(hdfs 权限，yarn配置)，版本不兼容(使用cdh中的hadoop运行flink)，或者其他的错误。

日志文件

在某种情况下，flink yarn session 部署失败是由于它自身的原因，用户必须依赖于yarn的日志来进行分析。最有用的就是[yarn log aggregation][] 。启动它，用户必须在yarn-site.xml文件中设置yarn.log-aggregation-enable 属性为true。一旦启用了，用户可以通过下面的命令来查看一个失败的yarn session的所有详细日志。

yarn logs -applicationId <application ID>

yarn client控制台和web界面

flink yarn client也会打印一些错误信息在控制台上，如果错误发生在运行时(例如如果一个taskmanager停止工作了一段时间)

除此之外，yarn resource manager的web界面(默认端口是8088)。resource manager的端口是通过yarn.resourcemanager.webapp.address参数来配置的。

它运行在yarn 程序运行的时候查看日志和程序失败的时候查看日志用户查找问题。

### 针对指定的hadoop版本构建yarn client ###

用户可以使用hadoop发行版。例如，hortonworks，CDH或者MapR等版本去构建 flink。请参考[构建指南][Link 4]获取详细信息

## ##

### 在yarn上运行flink使用防火墙 ###

一些yarn 集群使用防火墙来控制集群的网络和其他网络的通信。在这种设置下，flink只能通过集群的网络来提交任务到yarn session。针对生产环境下使用是不可行的，flink允许配置所有相关服务的端口范围，通过这些端口范围的配置，用户也可以透过防火墙来提交flink job。

目前，两个服务都需要提交任务：

*  jobmanager(yarn中的applicationMaster)
 *  jobmanager内部运行的blobserver

当向flink提交一个任务的时候，blobserver将会把用户的代码分发到所有工作节点(taskManagers)。jobmanager接收任务本身，并触发执行。

以下两个配置参数可以指定端口：

*  yarn.application-master.port
 *  blob.server.port

这两个配置选项接收单一的端口(例如："50010")，区间("50000-50025")，或者同时指定多个("50010,50011,50020-50025,50050-50075")。

(hadoop也使用的是类似的机制，例如：yarn.app.mapreduce.am.job.client.port-range)

### flink on yarn 内部实现 ###

本节主要描述flink和yarn是如何交互的

![70][]

YARN 客户端需要访问 Hadoop 配置，从而连接 YARN 资源管理器和 HDFS。可以使用下面的策略来决定 Hadoop 配置：

*  测试 YARN\_CONF\_DIR, HADOOP\_CONF\_DIR 或 HADOOP\_CONF\_PATH 环境变量是否设置了（按该顺序测试）。如果它们中有一个被设置了，那么它们就会用来读取配置。
 *  如果上面的策略失败了（如果正确安装了 YARN 的话，这不应该会发生），客户端会使用 HADOOP\_HOME 环境变量。如果该变量设置了，客户端会尝试访问 $HADOOP\_HOME/etc/hadoop (Hadoop 2) 和 $HADOOP\_HOME/conf(Hadoop 1)。

当启动一个新的 Flink YARN Client会话，客户端首先会检查所请求的资源（容器和内存）是否可用。之后，它会上传包含了 Flink 配置和 jar文件到 HDFS（步骤 1）。

客户端的下一步是请求（步骤 2）一个 YARN 容器启动 ApplicationMaster （步骤 3）。因为客户端将配置和jar 文件作为容器的资源注册了，所以运行在特定机器上的 YARN 的 NodeManager 会负责准备容器（例如，下载文件）。一旦这些完成了，ApplicationMaster (AM) 就启动了。

JobManager 和 AM 运行在同一个容器中。一旦它们成功地启动了，AM 知道 JobManager 的地址（它自己）。它会为 TaskManager 生成一个新的 Flink 配置文件（这样它们才能连上 JobManager）。该文件也同样会上传到 HDFS。另外，AM 容器同时提供了 Flink 的 Web 界面服务。Flink 用来提供服务的端口是由用户 + 应用程序 id 作为偏移配置的。这使得用户能够并行执行多个 Flink YARN 会话。

之后，AM 开始为 Flink 的 TaskManager 分配容器，这会从 HDFS 下载 jar 文件和修改过的配置文件。一旦这些步骤完成了，Flink 就安装完成并准备接受任务了。

[http_flink.apache.org_downloads.html]: http://flink.apache.org/downloads.html
[Center]: /images/20220519/88272878a1794284b2d3aa1a5eb0ba91.png
[Link 1]: http://flink.apache.org/faq.html#yarn-deployment
[Link 2]: https://ci.apache.org/projects/flink/flink-docs-release-1.4/ops/config.html
[Link 3]: https://ci.apache.org/projects/flink/flink-docs-release-1.4/ops/cli.html
[yarn log aggregation]: http://hortonworks.com/blog/simplifying-user-logs-management-and-access-in-yarn/
[Link 4]: https://ci.apache.org/projects/flink/flink-docs-release-1.4/start/building.html
[70]: /images/20220519/ab5580dd2eb14652bffed4c2074831be.png