flinkonyarn集群安装部署

flink集群安装部署

yarn集群模式

快速开始在yarn上启动一个一直运行的flink集群在yarn上运行一个flink jobflink yarn session启动flink session提交任务到flink在yarn上运行一个独立的flink job用户依赖jar包和classpathflink on yarn的故障恢复调试一个失败的yarn session日志文件yarn client控制台和web界面针对指定的hadoop版本构建yarn client在yarn上运行flink使用防火墙flink on yarn 内部实现

快速开始

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

启动一个yarn session使用4个taskmanager(每个节点4GB内存)

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

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

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

在yarn上运行一个flink job

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

flink yarn session

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

必须的依赖

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

如果你在使用flink yarn client的时候有什么问题

启动flink session

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

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

下载flink

下载hadoop2对应的flink安装包

使用下面命令解压：

tar xvzf flink-1.4.2-bin-hadoop2.tgz
cd flink-1.4.2/

启动一个session

使用下面命令启动一个session

./bin/yarn-session.sh

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

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

请注意：client必须要设置YARN_CONF_DIR或者HADOOP_CONF_DIR环境变量，通过这个环境变量来读取YARN和HDFS的配置信息。

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

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

该系统默认会使用这个配置文件：conf/flink-conf.yaml。如果你想修改一些参数

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 

附着到一个已存在的session

使用下面命令启动一个session

./bin/yarn-session.sh

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

用法:
   必须
     -id,--applicationId         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

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

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

用法: run [OPTIONS]  
  "run" 操作参数:
     -c,--class            如果没有在jar包中指定入口类，则需要在这里通过这个参数指定

     -m,--jobmanager       指定需要连接的jobmanager(主节点)地址
                                      使用这个参数可以指定一个不同于配置文件中的jobmanager

     -p,--parallelism    指定程序的并行度。可以覆盖配置文件中的默认值。

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

例子：

wget -O LICENSE-2.0.txt http://www.apache.org/licenses/LICENSE-2.0.txt
hadoop fs -copyFromLocal LICENSE-2.0.txt hdfs:/// ...
./bin/flink run ./examples/batch/WordCount.jar \
        hdfs:///..../LICENSE-2.0.txt hdfs:///.../wordcount-result.txt

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

Exception in thread "main" org.apache.flink.compiler.CompilerException:
    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包在类路径中的位置可以通过下面参数来控制：

* ORDER: (默认) 添加jar包到系统类路径下面，按照字典顺序.
* FIRST: 将jar添加到类路径的前面.
* 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 

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。

在yarn上运行flink使用防火墙

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

目前，两个服务都需要提交任务：
jobmanager(yarn中的applicationMaster)jobmanager内部运行的blobserver

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

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

yarn.application-master.portblob.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是如何交互的

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 就安装完成并准备接受任务了。