Flink 1.13与Hadoop 3.x集成指南：版本兼容与依赖冲突排查

刚把公司的Flink任务从旧环境迁到Hadoop 3.x集群时我一度怀疑是集群配置有问题同一个WordCount本地IDEA跑得好好的一提交到集群就报NoClassDefFoundError: org/apache/hadoop/mapred/JobConf查了一圈才发现问题出在Flink 1.13和Hadoop 3.x的集成上。这个坑不算深但第一次踩到会很懵因为报错信息五花八门有的指向ClassNotFoundException有的是NoSuchMethodError还有的干脆是java.lang.IllegalArgumentException全都不带重样的。如果你正准备把Flink 1.13和Hadoop 3.x搭在一起不管是本地开发还是部署到集群这篇文章应该能帮你省下半天排查时间。我会从版本兼容的根本原因讲起给出可用的依赖配置走一遍完整的集成流程再把我遇到过的高频报错和修复方法整理成清单方便你对着抄。1. 先说结论Flink 1.13 和 Hadoop 3.x 的兼容关系到底卡在哪1.1 Flink 1.13 官方到底带了什么 Hadoop 依赖要搞清楚怎么解决先得知道为什么会有问题。Flink从1.11版本开始就做了一个重要的变更官方发布的二进制包不再捆绑Hadoop依赖。也就是说你从官网下载的flink-1.13.6-bin-scala_2.12.tgz解压后lib目录里默认是没有Hadoop相关jar包的。这个设计是有意为之。早期版本的Flink把Hadoop依赖直接内置进来虽然用户开箱即用但也带来两个麻烦一是发行包体积很大动辄几百MB二是Hadoop生态迭代快Flink内置的版本和用户集群里的Hadoop版本经常对不上容易在运行时出现各种莫名其妙的冲突。所以从1.11开始官方选择“解耦”——你要连HDFS、要提交到YARN、要用到Hadoop的输入输出格式就自己把Hadoop客户端依赖补上。但问题也出在这里Flink 1.13官方测试时主要基于Hadoop 2.x系列的client而很多公司这时候已经升级到了Hadoop 3.x。网上很多教程写的依赖还是老一套hadoop-client2.7.x、2.8.x照抄过来连HDFS都访问不了更别提交任务了。1.2 兼容性表象ClassNotFound 只是导火索很多人以为集成失败就是缺jar包补上就行。实际上Flink 1.13和Hadoop 3.x之间的兼容问题分三层第一层是客户端API的兼容。Hadoop 3.x相比2.x虽然保持了大部分API兼容但有些内部接口和类路径发生了变化。比如org.apache.hadoop.mapred.JobConf、org.apache.hadoop.fs.FileSystem这类核心类都在但有些辅助类被移动了位置或者改了签名如果Flink内部引用的是编译期绑定的旧类签名运行时就会报NoSuchMethodError。第二层是传输层协议的兼容。Hadoop 3.x的HDFS RPC协议和2.x不完全相同如果客户端是2.x、服务端是3.x某些情况下能连上但很多操作会超时或直接抛异常。反过来用3.x的client访问2.x的集群也会遇到协议协商失败的问题。所以理想的做法是client版本和集群版本保持一致或者至少在major版本上对齐。第三层是依赖冲突。这一层最常见也最让人头大。Hadoop 3.x引入了protobuf 3.x、guava 27.0-jre这些新版本依赖而Flink 1.13自身打包了一堆旧版本的第三方库比如flink-runtime里带的guava可能是18.0或20.0。两个版本的库同时出现在classpath上谁先加载谁就生效于是就会出现“代码没错、依赖也加了但运行时就是找不到某个方法”的情况。1.3 两条路线官方发行包换 jar vs 工程依赖引 client针对上面的问题社区里最常用的集成方案有两条路线你需要根据自己场景选路线A使用Flink官方提供的pre-bundled Hadoop uber jar。Flink在flink-shaded项目里有发布flink-shaded-hadoop-2-uber和flink-shaded-hadoop-3-uber这种“全家桶”jar把Hadoop client及其所有依赖打成一个包。你只需要把这个uber jar放进Flink的lib目录或者作为工程依赖引进来就能解决大部分classpath问题。缺点是包很大几十MB到上百MB而且因为把所有依赖都shade了排错时更难看出是哪个依赖冲突。路线B在工程里显式引入hadoop-client依赖并处理冲突。这是更标准、更适合生产环境的做法。你根据自己的Hadoop集群版本比如3.3.x在Maven或Gradle里引入对应的hadoop-client然后通过依赖排除、shade插件等方式解决冲突。缺点是配置要细心尤其是guava、protobuf这些老冤家。我个人的建议是本地开发和快速验证用路线A正式工程和长期维护用路线B。两条路线的具体配置下面会详细讲。2. Maven/Gradle 工程里把 Hadoop 3.x 依赖配稳2.1 标准的 Maven 依赖写法与版本选择假设你的Flink工程是基于Maven构建的最省心的一组依赖是这样的。注意我用的是provided作用域后面会解释为什么properties flink.version1.13.6/flink.version hadoop.version3.3.4/hadoop.version /properties dependencies !-- Flink 核心 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java_2.12/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients_2.12/artifactId version${flink.version}/version /dependency !-- Hadoop Client一定要加 -- dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version${hadoop.version}/version scopeprovided/scope /dependency !-- 如果要用HDFS可能需要额外引入 -- dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-hdfs/artifactId version${hadoop.version}/version scopeprovided/scope /dependency /dependencies这里hadoop.version的选择有个技巧尽量和你实际连接的Hadoop集群版本保持一致。如果公司集群是CDH或HDP发行版最好用发行版对应的版本号而不是无脑用Apache社区版。比如CDH 7.1底层是Hadoop 3.1.1你用3.3.x的client去连虽然大多数操作没问题但某些冷门接口可能协议不兼容。2.2 provided vs compile不同运行模式的 scope 取舍provided是很多新手容易忽略的点。这里我解释一下为什么用provided而不是默认的compile如果你的任务最终是提交到Flink集群不管是Standalone还是YARN集群的lib目录里已经有了Hadoop依赖那么业务jar里再打一份就是重复的还容易引发类加载冲突。所以用provided让业务jar不包含Hadoop依赖运行时从Flink集群里找。但如果你的任务是本地开发调试比如在IDEA里直接右键运行main方法IDEA默认会使用compile或runtime作用域的依赖provided的依赖在运行时也通常会被加进来IDEA的处理和Maven标准行为略有差异所以一般也能跑起来。有一种特殊情况如果你用flink run以“附加jar”的方式提交而且Flink集群的lib里没有Hadoop依赖那provided会导致运行时报ClassNotFoundException。这时候你需要临时把某个依赖改成compile并重打jar。我的建议是开发调试阶段用provided就行集群提交阶段要确认Flink的lib目录里确实有匹配的Hadoop jar。如果没有就参考第三节的方法先把lib目录配好而不是改scope把依赖塞进业务jar。2.3 冲突源头protobuf、guava、javax.annotation 逐个处理即使依赖配置写对了mvn package后一运行还是有概率报错。这类问题八成出在下面几个“惯犯”身上。protobufHadoop 2.x依赖protobuf 2.5.0Hadoop 3.x依赖protobuf 3.x。Flink 1.13自身的某些模块引用了protobuf 2.5.0或3.x取决于你用的Connector。如果classpath里同时出现两个大版本会看到类似java.lang.NoSuchMethodError: com.google.protobuf.LiteralByteString的错误。排查办法是在mvn dependency:tree里看protobuf从哪里冒出来的然后用exclusions把多余的那个排除掉。通常保留和Hadoop一致的那个版本。guavaFlink 1.13的flink-runtime里带了旧版guava比如18.0而Hadoop 3.x需要guava 27.0。如果旧版guava被先加载Hadoop内部调用新API时就会报NoSuchMethodError。解决方案有两种一是排除Flink自带的guava但可能有风险因为Flink本身也在用二是把Hadoop的guava打包时通过shade重定位让Hadoop使用自己的版本。如果嫌麻烦直接用flink-shaded-hadoop-3-uber更省事因为shaded包已经处理过这些问题。javax.annotationHadoop 3.x的一些注解类需要javax.annotation-api如果没引入会报ClassNotFoundException: javax.annotation.Nullable。这个简单加一个依赖就行dependency groupIdjavax.annotation/groupId artifactIdjavax.annotation-api/artifactId version1.3.2/version /dependencycommons-logging / log4jHadoop的日志依赖和Flink冲突也比较常见但通常不会阻断运行只是会在控制台刷一堆警告。如果想清理就在引入hadoop-client时排除org.slf4j:slf4j-log4j12之类的传递依赖dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version${hadoop.version}/version scopeprovided/scope exclusions exclusion groupIdorg.slf4j/groupId artifactIdslf4j-log4j12/artifactId /exclusion /exclusions /dependency2.4 非 Maven 场景怎么处理可能有人用的是Gradle或者干脆不是标准工程比如用Zeppelin、用IDE临时拼jar。Gradle的话思路一样只是语法换成Gradle的implementation/compileOnly推荐用compileOnly对应Maven的provideddependencies { implementation org.apache.flink:flink-java:1.13.6 implementation org.apache.flink:flink-streaming-java_2.12:1.13.6 compileOnly org.apache.hadoop:hadoop-client:3.3.4 }如果你的环境是Zeppelin这类的交互式环境没法改构建文件那就老老实实用路线A把flink-shaded-hadoop-3-uber扔进Flink的lib目录所有依赖问题一次性解决。3. 从开发到集群一套能跑通的集成实操流程3.1 本地 IDEA 调试时的配置项目拉下来以后先在本地跑通这是最快速验证依赖是否配对的场景。IDEA里需要注意两个地方第一确认provided依赖在运行时被包含。默认IDEA对provided的处理是“编译时包含、运行时也包含”但也有人改了设置导致运行时没有。如果你一运行就报NoClassDefFoundError先检查Run Configuration里有没有勾选Include provided dependencies相关的选项不同版本IDEA入口不一样或者临时把hadoop-client的scope改成compile跑通再说但记得改回来。第二设置HADOOP_HOME和HDFS配置。如果本地要直连HDFS需要能加载到core-site.xml和hdfs-site.xml。简单做法是在Run Configuration的Environment variables里加上HADOOP_HOME指向你的Hadoop安装目录Flink的Hadoop FileSystem实现会尝试从classpath读取配置。更保险的方式是直接在src/main/resources里放一份core-site.xml内容只包含你集群的fs.defaultFSconfiguration property namefs.defaultFS/name valuehdfs://namenode:8020/value /property /configuration这样无论是本地跑还是打成jar到集群都能找到NameNode地址。注意这种方式会把配置打进业务jar如果集群的配置文件不同可能被覆盖生产环境建议用HADOOP_CONF_DIR环境变量动态指定。3.2 集群模式Flink 安装目录的 jar 处理本地跑通之后要部署到Flink集群这步是最容易出问题的。无论你用的是Standalone集群还是YARN上的Flink Session核心思路都是一样的让Flink的lib目录里存在Hadoop 3.x的client jar。实操步骤以flink-1.13.6为例下载并解压Flink二进制包如果你还没有的话wget https://archive.apache.org/dist/flink/flink-1.13.6/flink-1.13.6-bin-scala_2.12.tgz tar -xzf flink-1.13.6-bin-scala_2.12.tgz cd flink-1.13.6查看lib目录下已有的Hadoop相关jar。如果是从官网下的干净包通常没有Hadoop jar。如果你之前用过别人做的集成平替包可能会有flink-shaded-hadoop-2-uber-*.jar这种Hadoop 2.x的uber jar和Hadoop 3.x集群不搭建议先删掉ls lib/ # 如果有 flink-shaded-hadoop-2-uber-xxx.jar rm lib/flink-shaded-hadoop-2-uber-xxx.jar下载Hadoop 3.x的client jar。这里可以是flink-shaded-hadoop-3-uber也可以是一组hadoop-common、hadoop-hdfs、hadoop-client等正规jar。最省事的是用shaded uber jar官方已经把所有传递依赖都打好了wget -P lib/ https://repo1.maven.org/maven2/org/apache/flink/flink-shaded-hadoop-3-uber/3.1.1.7.2.1.0-9-0/flink-shaded-hadoop-3-uber-3.1.1.7.2.1.0-9-0.jar如果你用的是CDH/华为等发行版也可以从发行版的maven仓库拉对应的Hadoop client不过手工处理依赖会很麻烦我个人还是推荐shaded方式。设置环境变量HADOOP_CLASSPATHYARN模式尤其需要。Flink在提交到YARN时会通过HADOOP_CLASSPATH去加载Hadoop相关类。你可以这样生成并验证export HADOOP_CLASSPATH$(hadoop classpath) echo $HADOOP_CLASSPATH | head -c 200如果全局环境没配好也可以在conf/flink-conf.yaml里指定env.java.opts: -Djava.library.path/usr/local/hadoop/lib/native不过这个不是必须的只有用HDFS native压缩时才需要。3.3 一个最小验证任务的完整过程配置完成后用一个小工具类验证集成是否成功。下面这段代码读取HDFS文件系统的根目录列表并统计文件数。这比跑完整WordCount更聚焦能直接验证HDFS连通性和Hadoop client的有效性import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import java.net.URI; public class HadoopConnectCheck { public static void main(String[] args) throws Exception { // 设置HDFS地址也可以从core-site.xml读取 String hdfsUri args.length 0 ? args[0] : hdfs://namenode:8020; Configuration conf new Configuration(); conf.set(fs.defaultFS, hdfsUri); FileSystem fs FileSystem.get(URI.create(hdfsUri), conf); FileStatus[] statuses fs.listStatus(new Path(/)); System.out.println(HDFS root files/dirs: statuses.length); for (FileStatus status : statuses) { System.out.println(status.getPath().toString()); } fs.close(); } }在IDEA里直接把hdfs://namenode:8020换成你的集群地址运行。如果控制台打印出目录列表说明Hadoop 3.x的client在你的Flink环境中已经能正常工作。打成jar提交到集群的命令也一并给出mvn clean package -DskipTests flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 2048 \ target/flink-hadoop3-demo-1.0-SNAPSHOT.jar hdfs://namenode:8020注意-m yarn-cluster只适用于Flink 1.13及之前的版本提交到YARN的模式如果你用的是Flink 1.15推荐改用-t yarn-per-job或-t yarn-application但本文不展开。3.4 部署方式差异Standalone / YARN / K8s不同的部署模式下Hadoop集成的侧重点不太一样这里分别说下经验Standalone模式Flink独立集群你在每台TaskManager和JobManager的flink/lib目录下都要放入相同的Hadoop client jar。因为任务运行时JobManager和TaskManager各自需要加载Hadoop类只放在一台机器上是没用的。建议用同一份Flink安装包的lib目录用脚本同步到所有节点。YARN模式Flink on YARN这时Flink的JobManager和TaskManager都是跑在YARN Container里的它们的classpath由Flink的lib目录决定所以还是要保证提交机上的flink/lib里有Hadoop client jar。另外YARN模式特别依赖HADOOP_CLASSPATH这个环境变量必须在提交flink run命令的那台机器上配置好因为Flink要用它去和YARN ResourceManager交互。K8s模式用Flink的官方Docker镜像时你需要自己构建一个包含Hadoop client的镜像。参考Dockerfile大致如下FROM flink:1.13.6-scala_2.12 # 下载hadoop client并在镜像中注册环境变量 ENV HADOOP_CLASSPATH/opt/flink/lib/hadoop/* COPY hadoop-jars/*.jar /opt/flink/lib/镜像做好后加上HADOOP_CLASSPATH环境变量K8s部署才能正常连HDFS。4. 高频报错的排查与修复经验集成过程中踩过的坑我整理成一个速查表后面再逐个展开说明。这张表适用于Flink 1.13 Hadoop 3.x的常见组合遇到问题先对号入座典型报错信息直接原因快速解法NoClassDefFoundError: org/apache/hadoop/mapred/JobConf缺少Hadoop client依赖确认lib目录或工程依赖里有Hadoop 3.x clientNoSuchMethodError: com.google.common.base.Preconditions.checkArgumentguava版本冲突排除Flink自带旧guava或使用shaded的Hadoop uber包NoClassDefFoundError: com/google/protobuf/LiteralByteStringprotobuf版本冲突统一protobuf大版本通常保留3.xClassNotFoundException: javax.annotation.Nullable缺失javax.annotation-api手动添加javax.annotation-api依赖FileSystem closed或Filesystem closed在Flink中使用HDFS连接后未正确关闭用FileSystem的close或在Flink算子中谨慎管理连接生命周期java.io.IOException: No FileSystem for scheme: hdfsFileSystem实现类未被加载确认hadoop-hdfs和hadoop-common在classpath中检查core-site.xml中fs.hdfs.impl配置org.apache.hadoop.security.AccessControlException: Permission deniedHDFS权限问题检查Kerberos身份或fs.defaultFS访问的用户必要时配置代理用户4.1 NoClassDefFoundError 与 NoSuchMethodError 的区分很多人看到NoClassDefFoundError和NoSuchMethodError就慌其实这里的排查逻辑很简单。NoClassDefFoundError代表某个类在编译期存在但运行期的classpath里找不到。结合Hadoop集成场景绝大多数情况是Hadoop jar缺失或路径不对。比如Flink内部要初始化HDFS文件系统结果org.apache.hadoop.hdfs.DistributedFileSystem不在classpath里就会抛这个错。解决办法就是补jar没有捷径。NoSuchMethodError代表类存在但方法签名对不上通常不是缺jar而是版本不对。比如Hadoop 3.x把自己用的guava升级到了27.0而你的classpath里先被加载了Flink自带的guava 18.0checkArgument的签名有所变化Hadoop内部调用时就找不到对应方法。这种情况最有效的定位方式是打开Maven的依赖树逐层排查。排查命令我建议记下来mvn dependency:tree -Dincludescom.google.guava:guava mvn dependency:tree -Dincludescom.google.protobuf:protobuf-java mvn dependency:tree -Dincludesorg.apache.hadoop:hadoop-client看输出结果里这些库分别被哪些构件引入有没有多个版本。如果有就在引入方上做exclusions排除或者统一用dependencyManagement锁定版本。4.2 protobuf 与 guava 的 NoSuchMethodError这两个是我见过最频繁的“元凶”单独拿出来讲。protobuf排查Hadoop 3.x自己带的是protobuf-java 3.5.0或更高版本但Flink 1.13里的某些connector比如Kafka可能会传递性地引入protobuf 2.5.0。一旦2.5.0在前Hadoop在序列化RPC消息时调用的新API就不存在。这个时候有两个处理办法在依赖里排除旧版本。比如flink-connector-kafka_2.12如果带了protobuf 2.5.0就把它exclude掉。用Maven的dependencyManagement统一protobuf版本到3.xdependencyManagement dependencies dependency groupIdcom.google.protobuf/groupId artifactIdprotobuf-java/artifactId version3.11.4/version /dependency /dependencies /dependencyManagementguava排查Hadoop 3.x基本要求guava 27.0。Flink 1.13在flink-runtime里捆绑了guava 18.0或20.0不同小版本略有差别。如果你的业务代码在同一个算子中同时调用了Flink的API和Hadoop的API两者对guava版本的诉求互相冲突就很尴尬。我的建议是除非你对类加载机制很熟悉否则别轻易去排除Flink自带的guava。最稳妥的方案是把Hadoop client替换为flink-shaded-hadoop-3-uber。shaded jar把所有依赖包括guava都重定位了比如com.google.common被改成了org.apache.flink.shaded.com.google.common从而绕开和Flink自带guava的冲突。4.3 权限和配置文件的小坑Hadoop 3.x在安全认证上比2.x严格不少即使你本地没有启用Kerberos也容易踩到权限问题。我遇到过两个比较典型的情况一是HDFS目录权限拒绝。代码里用fs.listStatus(new Path(/))时如果当前提交任务的用户对/没有读权限就会报AccessControlException。解决思路不是改HDFS权限而是查看你当前用户是谁。Flink on YARN模式下提交用户是执行flink run的系统用户Standalone模式下TaskManager进程是哪个用户启动的就会以哪个用户访问HDFS。你可以通过HADOOP_USER_NAME环境变量临时指定一个代理用户来测试export HADOOP_USER_NAMEhdfs flink run -m yarn-cluster ...不过这只是调试方法生产环境建议配置好Kerberos或Sentry/Ranger的权限策略。二是core-site.xml里的fs.hdfs.impl配置被忽略。某些发行版的Hadoop 3.x会要求core-site.xml显式指定fs.hdfs.impl为org.apache.hadoop.hdfs.DistributedFileSystem否则会出现No FileSystem for scheme: hdfs。如果你确认client jar都在但就是连不上可以检查集群提供的core-site.xml和hdfs-site.xml是否被加载。一个快速验证方式是在代码里手动加载配置Configuration conf new Configuration(); conf.addResource(new Path(/etc/hadoop/conf/core-site.xml)); conf.addResource(new Path(/etc/hadoop/conf/hdfs-site.xml));如果这样能跑通说明是classpath里配置文件缺失不是代码问题。4.4 时区、临时目录与其他“边角料”问题除了上面这些主流问题还有几个边角配置容易让人卡半天时钟偏差。Hadoop 3.x对时钟同步比较敏感JobManager和NameNode之间的时差超过某个阈值默认2分钟Kerberos的TGT就会失效然后报各种认证重启错误。如果你们机房NTP没配好这会是最耗时的排查方向之一。简单检查方法是登录JobManager所在的机器对比date和时间服务器。本地临时目录。Hadoop在落盘shuffle、写临时文件时会用到java.io.tmpdirFlink on YARN模式下这个目录可能在容器里空间很小。如果发现任务跑了一段时间后偶发性报Disk Out of Space或No space left on device优先检查/tmp目录挂载情况然后调整env.java.opts里的临时目录路径env.java.opts: -Djava.io.tmpdir/data/flink_tmpFlink的classloader.resolve-order配置。Flink 1.13里classloader.resolve-order默认是child-first也就是优先加载用户jar里的类。如果你用户jar里也打了Hadoop依赖而Flink lib里也有可能会出现“两套Hadoop类同时存在”的幺蛾子。这时可以把这个参数改成parent-first试试但要注意别把用户自己的其他特殊依赖也弄乱。4.5 顺手几个提高排查效率的小技巧最后再分享几个我自己的操作习惯遇到集成问题能省不少时间技巧一用flink info提前验证jar的类加载。Flink 1.13提供了flink info命令可以用来打印jar里的类结构虽然主要用途是检查JobGraph但对发现classpath里的重复类也有参考价值。技巧二善用-Dlog4j.configuration打开详细日志。Hadoop自己有一套日志输出如果你只开Flink的debug日志可能看不清Hadoop内部的握手和认证过程。可以在conf/log4j.properties里临时把org.apache.hadoop的级别调到DEBUG跑一次最小任务再调回去。日志多不要紧关键是能看到具体卡在哪个协议交互步骤。技巧三本地写单元测试用MiniDFSCluster模拟HDFS。Hadoop 3.x的测试包里有MiniDFSCluster可以在本地起一个进程内的迷你HDFS节点做集成测试非常方便。这样就不需要每次连测试集群也避免把本地开发和远程集群的配置混在一起。说实话Flink 1.13集成Hadoop 3.x的坑排查到后面你会发现大部分不是“兼容性”问题而是“依赖版本的统一”问题。类加载的顺序、guava和protobuf的版本、Hadoop client与集群版本是否对齐这些才是真正的深水区。我个人现在的习惯是正式工程里固定用Hadoop 3.3.x并配合flink-shaded-hadoop-3-uber做兜底同时也保留一套干净的hadoop-client依赖以支持细粒度控制。如果你也经常在这两种方案之间来回切换建议在工程里准备好两套profile用Maven的-P参数在打包时灵活切换能省掉很多重复改pom的时间。希望这篇基于实际踩坑记录整理的集成指南能帮你少走弯路。如果你在生产环境还遇到过这里没提到的奇怪报错欢迎在评论区把报错信息和你的解决方式贴出来经验这种东西多交流一次就能让更多人少踩一次。