欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  科技

flinkonyarn集群安装部署

程序员文章站 2022-04-23 10:29:33
flink集群安装部署 yarn集群模式 快速开始在yarn上启动一个一直运行的flink集群在yarn上运行一个flink jobflink yarn session启动flink...

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

flinkonyarn集群安装部署

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