概要描述

ArgoDB 5.x 之后的版本，提供了 Flink 对接 ArgoDB 的案例，本文介绍一下将 flink 部署在 YARN 的方法；

适用版本为 flink 1.17.x

详细说明

1. 准备 Flink 包
2. 修改配置文件
3. 准备 TDH-Client 客户端
4. 启动 Flink on YARN 并验证
5. 停止 Flink

准备 flink 包

Flink 官网下载：https://flink.apache.org/downloads/

Flink 1.17.2 下载连接：https://www.apache.org/dyn/closer.lua/flink/flink-1.17.2/flink-1.17.2-bin-scala_2.12.tgz

并解压 tar -zxvf flink-1.17.2-bin-scala_2.12.tgz

准备 TDH-Client

建议选择 Manager 自带客户端，具体参见连接：随产品包上传与 Manager 自带客户端的使用

由于是 TDH 9.3.1 的 YARN，此时需要使用 hadoop3，source 时需要添加 hadoop3 参数；

source /root/TDH-Client/init.sh y y hadoop3

修改配置文件

1、添加 jaas.conf 文件

cp /etc/yarn1/conf/jaas.conf flink-1.17.2/conf/

2、修改 flink-conf.yaml

vi flink-1.17.2/conf/flink-conf.yaml，添加如下内容：

security.kerberos.login.use-ticket-cache: true
security.kerberos.login.keytab: /etc/yarn1/conf/yarn.keytab
security.kerberos.login.principal: yarn/gts-argo52x-171@GTSARGODB5
security.kerberos.login.contexts: Client
classloader.check-leaked-classloader: false
# 适配 hadoop2 的 kerberos 认证
security.kerberos.token.provider.hadoopfs.renewer: yarn

其中 security.kerberos.login.principal 的值，通过一下命令获取：

修改之后如下图所示：

3、准备 TDH-Client

建议选择 Manager 自带客户端，具体参见连接：随产品包上传与 Manager 自带客户端的使用

由于是 TDH 9.3.1 的 YARN，此时需要使用 hadoop3

4、启动 Flink on YARN

source /root/TDH-Client/init.sh y y hadoop3
export HADOOP_CLASSPATH=hadoop classpath
export HADOOP_CONF_DIR=/etc/yarn1/conf/
./flink-1.17.2/bin/yarn-session.sh -d

输出这样即为成功部署：

2024-04-29 13:46:12,198 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - YARN application has been deployed successfully.
2024-04-29 13:46:12,199 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface gts-argo52x-173:40249 of application 'application_1714360461092_0042'.
JobManager Web Interface: http://gts-argo52x-173:40249

然后使用 红框 中的 url 即可访问 JobManager 页面；

提交 Flink 任务验证

./flink-1.17.2/bin/flink run ./examples/streaming/TopSpeedWindowing.jar

5、停止 Flink

echo "stop" | ./bin/yarn-session.sh -id application_1714360461092_0042

附录

1. 如果出现 认证成功，但是 DFSClient token renewer 报错的，可以检查一下是否 hadoop 版本不一致。

本文是提交到 YARN-9.3，基于 hadoop3 的，但是 NullPointerException: null 之后出现了 hadoop-hdfs-2.7.2-transwarp-6.2.2.jar:2.7.2-transwarp-6.2.2 说明是 hadoop2 版本，只需要将 hadoop 切换为 hadoop3 版本问题即可解决。

2024-04-29 13:28:59,080 INFO  org.apache.flink.runtime.security.token.hadoop.KerberosLoginProvider [] - Successfully logged into KDC
2024-04-29 13:28:59,177 ERROR org.apache.flink.runtime.security.token.hadoop.HadoopFSDelegationTokenProvider [] - Failed to obtain delegation token for DFS[DFSClient[clientName=DFSClient_NONMAPREDUCE_1809563642_1, ugi=yarn/gts-argo52x-171@GTSARGODB5 (auth:KERBEROS)]]
java.lang.NullPointerException: null
        at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.getDelegationToken(ClientNamenodeProtocolTranslatorPB.java:933) ~[hadoop-hdfs-2.7.2-transwarp-6.2.2.jar:2.7.2-transwarp-6.2.2]

2. 如果是 hadoop2 版本的，出现认证成功后 DFSClient token renewer 报错的

可以在 flink-conf.yaml 中增加一个参数，如下图：
security.kerberos.token.provider.hadoopfs.renewer: yarn