Flink部署支持三种模式:本地部署、Standalone部署、Flink on Yarn部署。
独立(Standalone)模式由Flink自身提供资源,无需其他框架,这种方式降低了和其他第三方资源框架的耦合性,独立性非常强。但Flink 是大数据计算框架,不是资源调度框架,这并不是它的强项;所以还是应该让专业的框架做专业的事,和其他资源调度框架集成更靠谱。
Flink on Yarn模式,把资源管理交给Yarn实现,计算机资源统一由Haoop Yarn管理,生产环境测试。
Yarn(yet another resource negotiator)是一个通用分布式资源管理系统和调度平台,为上层应用提供统一的资源管理和调度。在集群利用率、资源统一管理和数据共享等方面带来巨大好处.
1 基础环境
1.1 服务器环境
操作系统环境为CentOS 8。
1)配置规划集群节点间免密访问
参考相关章节或附录的指南配置,可以有效提供部署和管理效率。
2)配置JAVA环境
参考相关章节或附录的指南配置。
3)配置HDFS存储集群
如果需要与HDFS存储集群集成,则需要提前完成配置。
参考相关章节或附录的指南配置,并且Flink规划集群或设备可网络访问。
4)配置zookeeper集群
如果需要部署Standalone模式,则需要提前完成配置。参考相关章节或附录的指南配置。
5)配置 Yarn集群
如果需要部署Flink on Yarn模式,则需要提前完成配置。参考相关章节或附录的指南配置。
1.2 Flink软件基础配置
在本实践案例中,采用的Flink软件包版本 1.14.5,Hadoop的版本为3.2,Spark软件的根目录(SPARK_HOME)为/opt/flink/flink。
源码下载可以通过官方源和国内源两种方式下载,官方源再国外,下载速度慢,国内源采用清华大学的源,速度相对较快,但只保留最新版本。
Apache官方:https://archive.apache.org/dist/flink/
清华大学:https://mirrors.tuna.tsinghua.edu.cn/apache/flink/
下载软件包:
# wget https://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.14.5/flink-1.14.5-bin-scala_2.12.tgz
# tar -zxvf flink-1.14.5-bin-scala_2.12.tgz
解压软件包:
# tar -zxvf flink-1.14.5-bin-scala_2.12.tgz
# ln -s flink-1.14.5 flink
# ls flink/
bin conf examples lib LICENSE licenses log NOTICE opt plugins README.txt
2 本地模式
最简单的启动方式,其实是不搭建集群,直接本地模式启动。
2.1 配置部署
在本地模式下,不需要启动任何的进程,仅仅是使用本地线程来模拟 Flink 的进程,适用于测试开发调试等,不用更改任何配置信息,只需要保证 JDK8 安装正常即可。
1)启动命令
# /usr/local/flink/bin/start-cluster.sh
2)关闭命令
# /usr/local/flink/bin/stop-cluster.sh
2.2 测试验证
1)Flink启动
# /usr/local/flink/bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host lake01.
Starting taskexecutor daemon on host lake01
2)访问验证
通过<local主机地址>:8081打开
3)jps查看
# jps
3968 Jps
1941 NameNode
3685 TaskManagerRunner
2790 NodeManager
3418 StandaloneSessionClusterEntrypoint
2159 DataNode
4)执行官方用例WordCount
通过执行官方示例,可以看到flink任务运行成功
# /opt/flink/flink/bin/flink run /opt/flink/flink/examples/batch/WordCount.jar
Executing WordCount example with default input data set.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
Job has been submitted with JobID d6987ed5b263fc5e297d5be1e28b465a
Program execution finished
Job with JobID d6987ed5b263fc5e297d5be1e28b465a has finished.
Job Runtime: 373 ms
Accumulator Results:
- 843a1470cb2c3e3169dfb25bcda7369d (java.util.ArrayList) [170 elements]
(a,5)
(action,1)
(after,1)
(against,1)
……
观察Flink WebUI,如下图
2.3 问题-提示无法连接Yarn服务
一、问题描述
从flink on yarn模式切换为本地模式,执行start-cluster.sh提示如下错误:
2022-10-13 20:38:05,757 WARN org.apache.flink.yarn.configuration.YarnLogConfigUtil [] - The configuration directory ('/opt/flink/flink-1.14.5/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
2022-10-13 20:38:05,946 INFO org.apache.hadoop.yarn.client.RMProxy [] - Connecting to ResourceManager at **lake02/******:8032
2022-10-13 20:38:06,017 INFO org.apache.flink.yarn.YarnClusterDescriptor [] - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2022-10-13 20:38:07,043 INFO org.apache.hadoop.ipc.Client [] - Retrying connect to server: **lake02/******:8032. Already tried 0 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=10, sleepTime=1000 MILLISECONDS)
……
2022-10-13 20:38:16,052 WARN org.apache.hadoop.ipc.Client [] - Failed to connect to server: **lake02/******:8032: retries get failed due to exceeded maximum allowed retries number: 10
java.net.ConnectException: Connection refused
二、问题分析
通过jps发现缺少TaskManagerRunner。
# jps
1902910 StandaloneSessionClusterEntrypoint
1861160 RunJar
1903349 SqlClient
1861350 RunJar
128746 NameNode
200744 QuorumPeerMain
1865793 Kafka
1911504 Jps
发现workers和masters文件均为空
三、解决方案
恢复masters和workers的内容
# cat masters
localhost:8081# cat workers
localhost
3 Standalone模式
3.1 概述
Standalone模式是最简单的一种集群模式,不需要Yarn、mesos等资源调度平台,自带集群,资源管理由flink集群管理,开发环境测试使用。
Standalone模式是一种主从模式,主要有两个组件构成分别是JobManager(Master)和TaskManager(Slave)。
当一个应用提交执行时,Flink的各个组件是如何交互协作的:
1)App程序通过rest接口提交给Dispatcher(rest接口是跨平台,并且可以直接穿过防火墙,不需考虑拦截)。
2)Dispatcher把JobManager进程启动,把应用交给JobManager。
3)JobManager拿到应用后,向ResourceManager申请资源(slots),ResouceManager会启动对应的TaskManager进程,TaskManager空闲的slots会向ResourceManager注册。
4)ResourceManager会根据JobManager申请的资源数量,向TaskManager发出指令(这些slots由你提供给JobManager)。
5)接着,TaskManager可以直接和JobManager通信了(它们之间会有心跳包的连接),TaskManager向JobManager提供slots,JobManager向TaskManager分配在slots中执行的任务。
6)最后,在执行任务过程中,不同的TaskManager会有数据之间的交换。
3.2 配置部署
一、节点规划
本次通过虚拟机部署,采用5个节点 ,每一个节点提供一块500GB的硬盘。
Hostname | IP | 用途 | 说明 |
labnode01 | 192.168.80.131 | master, jobmanager | OS:centos8.0 |
labnode02 | 192.168.80.132 | slave、taskmanager | OS:centos8.0 |
labnode03 | 192.168.80.133 | slave、taskmanager | OS:centos8.0 |
二、修改配置文件
1)修改flink-conf.yaml配置文件:
##配置master节点ip
jobmanager.rpc.address: 192.168.1.100##配置每个节点的可用slot,1 核CPU对应 1 slot
##the number of available CPUs per machine
taskmanager.numberOfTaskSlots: 30##默认并行度 1 slot资源
parallelism.default: 1
2)修改master和work配置文件
Master文件
# cat masters
labnode01:8081
workers文件
labnode02
labnode03
将以上文件分发各节点对应文件夹。
三、集群启动和关闭
在master节点上执行此脚本,就可以启动集群,前提要保证master节点到slaver节点可以免密登录。
因为它的启动过程是:先在master节点启动jobmanager进程,然后ssh到各slaver节点启动taskmanager进程。
启动集群
# /usr/local/flink/bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host lake02.
Starting taskexecutor daemon on host lake03.
Starting taskexecutor daemon on host lake04.
Starting taskexecutor daemon on host slake05.
停止集群:
# /usr/local/flink/bin/stop-cluster.sh
3.3 运行验证
1)启动Flink
# /usr/local/flink/bin/start-cluster.sh
2)访问flink webUI
3)执行官方用例WordCount
执行命令:
# /usr/local/flink/bin/flink run /usr/local/flink/examples/batch/WordCount.jar
……
- f27663f6191a378629eea720a988cc53 (java.util.ArrayList) [170 elements](a,5)
(action,1)
(after,1)
(against,1)
(all,2)
(and,12)
(arms,1)
……
查看Flink WebUI
4 Flink On Yarn模式
4.1 概述
独立(Standalone)模式由 Flink 自身提供资源,无需其他框架,这种方式降低了和其他第三方资源框架的耦合性,独立性非常强。但Flink 是大数据计算框架,不是资源调度框架,这并不是它的强项;所以还是应该让专业的框架做专业的事,和其他资源调度框架集成更靠谱。
在目前大数据生态中,国内应用最为广泛的资源管理平台是Yarn。Yarn(yet another resource negotiator)是一个通用分布式资源管理系统和调度平台,为上层应用提供统一的资源管理和调度。在集群利用率、资源统一管理和数据共享等方面带来巨大好处。
Flink on Yarn 企业生产环境运行Flink任务大多数的选择。
在强大的Yarn平台上,Flink是如何在Yarn上集成部署的,其过程是:客户端把Flink 应用提交给Yarn的ResourceManager,Yarn的ResourceManager会向Yarn的 NodeManager 申请容器。在这些容器上,Flink会部署JobManager 和 TaskManager 的实例,从而启动集群。Flink会根据运行在JobManger上的作业所需要的 Slot 数量动态分配TaskManager资源。
1)提交App之前,先上传Flink的Jar包和配置到HDFS,以便JobManager和TaskManager共享HDFS的数据。
2)客户端向ResourceManager提交Job,ResouceManager接到请求后,先分配container资源,然后通知NodeManager启动ApplicationMaster。
3)ApplicationMaster会加载HDFS的配置,启动对应的JobManager,然后JobManager会分析当前的作业图,将它转化成执行图(包含了所有可以并发执行的任务),从而知道当前需要的具体资源。
4)接着,JobManager会向ResourceManager申请资源,ResouceManager接到请求后,继续分配container资源,然后通知ApplictaionMaster启动更多的TaskManager(先分配好container资源,再启动TaskManager)。container在启动TaskManager时也会从HDFS加载数据。
5)最后,TaskManager启动后,会向JobManager发送心跳包。JobManager向TaskManager分配任务。
Flink提供了yarn上运行的3模式,分别为Session-Cluster,Application Mode和Per-Job-Cluster模式。
4.2 配置部署
一、节点规划
本次通过虚拟机部署,采用5个节点 ,每一个节点提供一块500GB的硬盘。
Hostname | IP | 用途 | 说明 |
labnode01 | 192.168.80.131 | master, jobmanager | |
labnode02 | 192.168.80.132 | slave、taskmanager | |
labnode03 | 192.168.80.133 | slave、taskmanager |
二、Yarn环境配置
在Yarn-site.xml中配置关闭内存校验。
Yarn-site.xml是hadoop中/etc/hadoop下的配置文件,否则flink任务可能会因为内存超标而被Yarn集群主动杀死。
<!-- Mem Check Start -->
<!-- 设置不检查虚拟内存的值,不然内存不够会报错 --><property><name>yarn.nodemanager.pmem-check-enabled</name><value>false</value></property><property><name>yarn.nodemanager.vmem-check-enabled</name><value>false</value></property><!-- Mem Check end -->
将修改后的配置文件分发到各节点,然后重启Yarn集群。
三、将Flink软件和配置文件分发到Flink集群规划节点
将Flink的配置文件conf/flink-conf.yaml恢复为初始状态。
4.3 Session-Cluster模式(yarn-session)
4.3.1 概述
Session-Cluster模式需要先启动Flink集群,向Yarn申请资源。以后提交任务都向这里提交。这个Flink集群会常驻在yarn集群中,除非手工停止。
在向Flink集群提交Job的时候, 如果资源被用完了,则新的Job不能正常提交.
缺点: 如果提交的作业中有长时间执行的大作业, 占用了该Flink集群的所有资源, 则后续无法提交新的job.
所以, Session-Cluster适合那些需要频繁提交的多个小Job, 并且执行时间都不长的Job。
会话模式有两种操作模式:
- 附加模式(默认):yarn-session.sh客户端将 Flink 集群提交给 YARN,但客户端一直在运行,跟踪集群的状态。如果集群失败,客户端将显示错误。如果客户端被终止,它也会发出集群关闭的信号。
- 分离模式(-d或--detached):yarn-session.sh客户端将 Flink 集群提交给 YARN,然后客户端返回。需要再次调用客户端或 YARN 工具来停止 Flink 集群。
4.3.2 常用命令
1)yarn-session.sh参数说明
使用bin/yarn-session.sh --help 查看可用参数:
Usage:Optional-at,--applicationType <arg> Set a custom application type for the application on Yarn-D <property=value> use value for given property-d,--detached If present, runs the job in detached mode-h,--help Help for the Yarn session CLI.-id,--applicationId <arg> Attach to running Yarn session-j,--jar <arg> Path to Flink jar file-jm,--jobManagerMemory <arg> Memory for JobManager Container with optional unit (default: MB)-m,--jobmanager <arg> Set to Yarn-cluster to use Yarn execution mode.-nl,--nodeLabel <arg> Specify Yarn node label for the Yarn application-nm,--name <arg> Set a custom name for the application on Yarn-q,--query Display available Yarn resources (memory, cores)-qu,--queue <arg> Specify Yarn queue.-s,--slots <arg> Number of slots per TaskManager-t,--ship <arg> Ship files in the specified directory (t for transfer)-tm,--taskManagerMemory <arg> Memory per TaskManager Container with optional unit (default: MB)-yd,--Yarndetached If present, runs the job in detached mode (deprecated; use non-Yarn specific option instead)-z,--zookeeperNamespace <arg> Namespace to create the Zookeeper sub-paths for high availability mode
2)启动命令
使用Yarn-session.sh命令申请资源初始化一个Flink集群,命令格式如下:
bin/yarn-session.sh <参数>
如示例:
# /opt/flink/flink/bin/yarn-session.sh -d
3)关闭Flink
停止 flink on Yarn 会话模式中的flink集群
yarn application -kill <appid>
或
echo "stop" | ./bin/flink -id <appid>
如示例:
# echo "stop" | /opt/flink/flink/bin/flink -id application_1661480406159_0025
4.3.3 运行验证
1)启动Flink
# /opt/flink/flink/bin/yarn-session.sh -d
执行结果:
访问Yarn WebUI:
访问Flink WebUI,http://lake04:38347
2)运行官方用例WordCount
# /opt/flink/flink/bin/flink run /opt/flink/flink/examples/batch/WordCount.jar
命令行执行结果:
Flink WebUI的首页:
Flink WebUI中的结果:
3)关闭
执行命令关闭Flink
# echo "stop" | /opt/flink/flink/bin/flink -id application_1661480406159_0025
4.4 Per-Job-Cluster模式(yarn-cluster)
4.4.1 概述
一个Job会对应一个Flink集群,每提交一个作业会根据自身的情况,都会单独向yarn申请资源,直到作业执行完成,一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享Dispatcher和ResourceManager,按需接受资源申请;适合规模大长时间运行的作业。
每次提交都会创建一个新的flink集群,任务之间互相独立,互不影响,方便管理。任务执行完成之后创建的集群也会消失。
4.4.2 常用命令
1)参数说明
flink run -m yarn-cluster --help;可用参数:
该模式下不需要先启动 yarn-session,确保 Hadoop 集群是健康的情况下直接提交 Job 命令:
bin/flink -m yarn-cluster <参数> <jar file>
如示例:
# /opt/flink/flink/bin/flink run -m yarn-cluster -yjm 1024 -ytm 1024 /opt/flink/flink/examples/batch/WordCount.jar
4.4.3 运行验证
1)启动并执行官方用例WordCount
# /opt/flink/flink/bin/flink run -m yarn-cluster -yjm 1024 -ytm 1024 /opt/flink/flink/examples/batch/WordCount.jar
执行结果:
访问Yarn WebUI:
4.4.4 优缺点
优点:随到随用,只有任务需要运行时才会开启flink集群;运行完就关闭释放资源,资源利用更合理;
缺点:对于小作业不太友好,
适用场景:适合大作业,长时间运行的大作业。
4.5 Application Mode
4.5.1 概述
Application Mode会在Yarn上启动集群, 应用jar包的main函数(用户类的main函数)将会在JobManager上执行. 只要应用程序执行结束, Flink集群会马上被关闭. 也可以手动停止集群.
与Per-Job-Cluster的区别: 就是Application Mode下, 用户的main函数式在集群中执行的
官方建议:
出于生产的需求, 我们建议使用Per-job or Application Mode,因为他们给应用提供了更好的隔离!
4.5.2 运行验证
启动
# /opt/flink/flink/bin/flink run-application -t yarn-application /opt/flink/flink/examples/batch/WordCount.jar
执行结果:
访问Yarn WebUI:
4.5.3 常见问题
任务提示 Could not allocate the required slot within slot request tim
一、错误日志
Caused by: java.util.concurrent.CompletionException: java.util.concurrent.CompletionException: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Slot request bulk is not fulfillable! Could not allocate the required slot within slot request timeoutat org.apache.flink.runtime.scheduler.DefaultScheduler.lambda$assignResource$8(DefaultScheduler.java:539)... 37 more
Caused by: java.util.concurrent.CompletionException: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Slot request bulk is not fulfillable! Could not allocate the required slot within slot request timeoutat java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308)at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:607)at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)... 35 more
Caused by: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Slot request bulk is not fulfillable! Could not allocate the required slot within slot request timeoutat org.apache.flink.runtime.jobmaster.slotpool.PhysicalSlotRequestBulkCheckerImpl.lambda$schedulePendingRequestBulkWithTimestampCheck$0(PhysicalSlotRequestBulkCheckerImpl.java:86)... 28 more
Caused by: java.util.concurrent.TimeoutException: Timeout has occurred: 300000 ms... 29 more
解决方案
将flink的配置文件conf/flink-conf.yaml恢复为初始状态,重新启动flink的Yarn session。
问题FLINK Could not get job jar and dependencies from JAR file: JAR file does not exist:
一、问题描述
使用flink客户端将执行flink提交到Yarn,输入-yjm参数提示错误
# /opt/flink/flink/bin/flink run -m Yarn-cluster -yn 2 -yjm 1024 -ytm 1024 /opt/flink/flink/examples/batch/WordCount.jar
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/flink/flink-1.14.5/lib/log4j-slf4j-impl-2.17.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/hadoop/hadoop-3.2.4/share/hadoop/common/lib/slf4j-reload4j-1.7.35.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
Could not get job jar and dependencies from JAR file: JAR file does not exist: -yn
二、问题分析
flink1.8版本之后已弃用该参数,ResourceManager将自动启动所需的尽可能多的容器,以满足作业请求的并行性。
三、解决方案
去掉即可
Deployment took more than 60 seconds. Please check if the requested resources are
一、问题描述
日志信息如下:
INFO org.apache.flink.yarn.YarnClusterDescriptor [] - Deployment took more than 60 seconds. Please check if the requested resources are available in the YARN cluster
INFO org.apache.flink.yarn.YarnClusterDescriptor [] - Deployment took more than 60 seconds. Please check if the requested resources are available in the YARN cluster
截图如下:
二、解决方案
配置yarn-site.xml
<property><name>yarn.scheduler.minimum-allocation-mb</name><value>1024</value></property><property><name>yarn.scheduler.maximum-allocation-mb</name><value>102400</value></property><property><name>yarn.nodemanager.resource.cpu-vcores</name><value>32</value></property><property><name>yarn.nodemanager.resource.memory-mb</name><value>51200</value></property>
Flink读取Hudi表时报错lassNotFoundException: *mapred.FileInputFormat
一、问题现象:
执行“select * from t1;”报错,报错信息如下:
Flink SQL> select * from t1;
……
[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: org.apache.hadoop.mapred.FileInputFormat
二、原因分析
mapred代表的是hadoop旧API,而mapreduce代表的是hadoop新的API
三、解决办法
解决办法为复制集群的hadoop-mapreduce-client-core.jar到Flink/lib中。
读取数据表失败NoSuchMethodError: *Preconditions.checkArgument
一、问题描述
创建表格式如下
CREATE TABLE t2(
uuid VARCHAR(20),
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
`partition` VARCHAR(20)
)
PARTITIONED BY (`partition`)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://10.101.12.140:9000/datas/flink-hudi/test0907/t1',
'table.type' = 'MERGE_ON_READ',
'read.tasks' = '1',
'read.streaming.enabled' = 'true',
'read.streaming.start-commit' = '20210316134557',
'read.streaming.check-interval' = '4'
);
INSERT INTO t2 VALUES('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'),('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'),('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'),('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'),('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'),('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'),('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'),('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');
执行成功,然后执行表内容查询
select * from t2;
报出如下错误:
Flink SQL> select * from t2;
[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputForm
Flink SQL>
二、原因分析
mapred代表的是hadoop旧API,而mapreduce代表的是hadoop新的API
三、解决办法
解决办法为复制集群的hive-exec-3.1.3.jar到各节点的flink/lib中。(注意hive-exec和hadoop版本的匹配)
启动失败NoSuchMethodError: *Preconditions.checkArgument
一、错误描述
通过bin/yarn-session.sh -d启动yarn-session失败,报错信息如下:
The program finished with the following exception:org.apache.hadoop.yarn.exceptions.YarnRuntimeException: java.lang.reflect.InvocationTargetExceptionat org.apache.hadoop.yarn.factories.impl.pb.RpcClientFactoryPBImpl.getClient(RpcClientFactoryPBImpl.java:81)at org.apache.hadoop.yarn.ipc.HadoopYarnProtoRPC.getProxy(HadoopYarnProtoRPC.java:48)……... 21 more
Caused by: java.lang.NoSuchMethodError: com.google.common.base.Preconditions.checkArgument(ZLjava/lang/String;Ljava/lang/Object;)Vat org.apache.hadoop.conf.Configuration.set(Configuration.java:1357)……at org.apache.hadoop.yarn.api.impl.pb.client.ApplicationClientProtocolPBClientImpl.<init>(ApplicationClientProtocolPBClientImpl.java:209)... 26 more
二、错误原因
Preconditions是guava下的工具类,hudi的源码依赖了不同的项目,这些项目使用了不同的guava版本,所报错误是由于运行时guava版本过旧,没有相应的方法。
三、解决方案
在HADOOP_HOME下查询hadoop使用的guava版本,将其拷贝到FLINK_HOME/lib下:
# find ./ -name guava*
./share/hadoop/common/lib/guava-27.0-jre.jar
./share/hadoop/hdfs/lib/guava-27.0-jre.jar
将文件复制到所有yarn集群的FLINK_HOME/lib下。
重新执行bin/yarn-session.sh -d,成功。
通过yarn启动flink失败-连接yarn失败
通过yarn模式启动flink,报出如下异常,关键信息如下:
# /usr/local/flink/bin/yarn-session.sh -d
….
2022-10-27 11:04:30,332 INFO org.apache.hadoop.yarn.client.RMProxy [] - Connecting to ResourceManager at /0.0.0.0:8032
…
2022-10-27 11:04:41,153 WARN org.apache.hadoop.ipc.Client [] - Retrying connect to server: 0.0.0.0/0.0.0.0:8032. Already tried 9 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=10, sleepTime=1000 MILLISECONDS)
2022-10-27 11:04:41,153 WARN org.apache.hadoop.ipc.Client [] - Failed to connect to server: 0.0.0.0/0.0.0.0:8032: retries get failed due to exceeded maximum allowed retries number: 10
java.net.ConnectException: Connection refused
……
原因分析:
1)检查是否启动hadoop集群, 如果没有启动, 是无法连接到hadoop的yarn。
2)flink运行于yarn上,flink要能找到hadoop配置,因为要连接到yarn的resourcemanager和hdfs。
如果正常启动还无法连接yarn, 可以查看一下hadoop的环境变量是否配置好。
在本实例中,时因为无法获取HADOOP_CONF_DIR的配置信息导致问题发生。
二、解决方案
设置HADOOP_CONF_DIR环境变量,并使之生效。
# cat /etc/profile | grep HADOOP_CONF_DIR
export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
# source /etc/profile
然后重新启动flink。
BTW:如果已经设置HADOOP_CONF_DIR环境变量,可能由于某种原因HADOOP_CONF_DIR环境变量没有生效,这个原因有很多。
5 参考资料
[01] https://blog.csdn.net/Vector97/article/details/117398947
[02] https://www.jianshu.com/p/8c9c897ea72a