flink集群安装部署

yarn集群模式

 

Flink入门及实战-上:

http://edu.51cto.com/sd/07245 <http://edu.51cto.com/sd/07245>

Flink入门及实战-下:

http://edu.51cto.com/sd/5845e <http://edu.51cto.com/sd/5845e>

 

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

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

启动一个yarn session使用4个taskmanager(每个节点4GB内存)
注意:如果自己的虚拟机没有这么大内存的话,可以吧-n设置小一点,对应的后面的内存-jm -tm也设置小一点,否则,如果内存不够用,会导致启动失败。
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.2
* hdfs(或者是其它支持hadoop的分布式文件系统)
如果你在使用flink yarn client的时候有什么问题,可以到这里查找答案
<http://flink.apache.org/faq.html#yarn-deployment>。

启动flink session

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

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

下载flink

下载hadoop2对应的flink安装包,点此下载 <http://flink.apache.org/downloads.html>。它包含了必须的文件。

使用下面命令解压:
tar xvzf flink-1.4.2-bin-hadoop2.tgz cd flink-1.4.2/
启动一个session

使用下面命令启动一个session
./bin/yarn-session.sh
这个命令将会输出下面内容:
用法: 必选 -n,--container <arg> 分配多少个yarn容器 (=taskmanager的数量) 可选 -D <arg> 动态属性
-d,--detached 独立运行 -jm,--jobManagerMemory <arg> JobManager的内存 [in MB]
-nm,--name 在YARN上为一个自定义的应用设置一个名字 -q,--query 显示yarn中可用的资源 (内存, cpu核数)
-qu,--queue <arg> 指定YARN队列. -s,--slots <arg> 每个TaskManager使用的slots数量
-tm,--taskManagerMemory <arg> 每个TaskManager的内存 [in MB] -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。如果你想修改一些参数,请查看我们的配置指南
<https://ci.apache.org/projects/flink/flink-docs-release-1.4/ops/config.html>。

 

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
执行这个命令将会显示下面内容:
用法: 必须 -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
请参考客户端命令行操作文档
<https://ci.apache.org/projects/flink/flink-docs-release-1.4/ops/cli.html>

 

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

"run" 参数可以编译和运行一个程序
用法: run [OPTIONS] <jar-file> <arguments> "run" 操作参数: -c,--class <classname>
如果没有在jar包中指定入口类,则需要在这里通过这个参数指定 -m,--jobmanager <host:port>
指定需要连接的jobmanager(主节点)地址 使用这个参数可以指定一个不同于配置文件中的jobmanager -p,--parallelism
<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
查看flink源码发现,wordCount.jar可以不提供参数,或者提供参数,提供参数的时候需要使用input和output参数指定:
上面的命令需要修改为如下格式才能正常执行[传递的两个参数需要使用-input和 -output来指定] ./bin/flink run
./examples/batch/WordCount.jar -input hdfs://hostname:port/LICENSE-2.0.txt
-output hdfs://hostname:port/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
<http://hortonworks.com/blog/simplifying-user-logs-management-and-access-in-yarn/>
 。启动它,用户必须在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。请参考构建指南
<https://ci.apache.org/projects/flink/flink-docs-release-1.4/start/building.html>
获取详细信息

 

在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是如何交互的



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

 

 

 

获取更多大数据资料,视频以及技术交流请加群:

QQ群号1:295505811(已满)

QQ群号2:54902210

QQ群号3:555684318