学习Hadoop如果只是在Shell里敲命令你可能很难真正理解这套体系在处理实际任务时的边界在哪里。我自己是在把HDFS和MapReduce都跑过一遍之后才专门花了一轮时间把Hadoop的Java API完整过了一遍。结果发现Shell能做的事用Java API都能做但反过来却不一定比如你希望在一个业务系统里把文件自动上传到HDFS、定时触发一轮MapReduce计算、或者把集群的状态集成到自己的管理平台上这些场景靠手工敲命令是撑不住的。所以这篇文章不是泛泛介绍Hadoop有哪些类而是围绕Hadoop Java API的使用这个主题把我在实际学习过程中踩过的坑、反复验证过的写法以及最后的代码模板都摊开来聊。核心内容就是两块HDFS客户端API和MapReduce作业API。前者负责文件层面的操作后者负责分布式计算任务的编写与提交。适合刚把Hadoop环境搭起来、想从脚本操作进阶到代码开发的人也适合那些在业务系统里需要嵌入Hadoop能力但还没找到合适切入点的人。1. 环境准备把依赖和配置一次理顺1.1 先搞清楚客户端角色很多初学者容易误解一点在你的开发机上引入Hadoop相关依赖不等于是装了一个Hadoop客户端服务。实际运行起来你的Java进程只是一个客户端它通过网络去连接独立的NameNode和DataNode进程。也就是说你的开发机上完全可以没有Hadoop安装包只要你的代码里带上正确的依赖、手上有集群的配置文件就能正常操作集群。我当时的实验环境是三节点集群Hadoop版本用的3.3.x。开发机是普通的Linux虚拟机网络可以和集群互通。开发机上没有部署任何DataNode或NameNode只保留了一份和集群保持一致的core-site.xml和hdfs-site.xml。这一步很关键因为后面所有FileSystem操作能不能找到NameNode都取决于这两份配置是否放进了classpath。1.2 Maven依赖必须这样写Hadoop的Java API依赖关系不算干净容易踩版本冲突。最省心的做法是引入hadoop-client这个聚合依赖。它能把你平时常用的HDFS客户端、YARN客户端、MapReduce客户端都带进来而且各依赖的版本已经对齐过比自己一个个引入hadoop-common、hadoop-hdfs要稳得多。dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.3.4/version scopeprovided/scope /dependency这里我把scope设置为provided这是第一个重要的习惯。因为当你把作业提交到集群时集群的HADOOP_CLASSPATH里已经有完整的一套Hadoop依赖了不需要再把它们打进自己的业务jar里。你强行打进去反而容易出现依赖版本不匹配比如集群里Hadoop 3.3.x你的jar里却带着Hadoop 2.x的类运行时候各种NoSuchMethodError会跑出来折磨人。如果你的代码里还需要用到一些Hadoop生态外的工具类比如commons-cli、commons-logging也不要手动指定版本让hadoop-client的传递依赖统一管理。实在需要排除某个冲突依赖时用mvn dependency:tree看完整依赖树再动手不要上来就凭感觉排除。1.3 配置文件加载机制别想当然Hadoop客户端读取配置的核心是Configuration类。它默认会从classpath下寻找core-site.xml、hdfs-site.xml、mapred-site.xml等文件按顺序加载后合并成一份配置。这个机制的好处是如果你的开发环境里已经放了这些配置文件代码里什么都不用写直接new Configuration()就能拿到集群地址。如果你没有把配置文件放进classpath也可以通过代码设置属性比如Configuration conf new Configuration(); conf.set(fs.defaultFS, hdfs://hadoop-master:8020);但我强烈不建议在业务代码里硬编码集群地址。因为你提交到生产环境之后可能面对的是另一个集群地址或者开启了HA的NameNode地址列表。硬编码会让代码的可移植性变得极差。最优雅的用法是本地开发时把集群的配置文件放在src/main/resources下打包后配置跟着进jar如果担心把集群真实配置打进分发jar不安全就运行时通过-Dhadoop.conf.dir/path/to/conf把外部配置目录加进JVM的classpath。另外要注意Configuration的加载是有优先级的classpath下同名的配置文件只会加载第一个。如果项目里同时存在多份core-site.xml很容易出现“明明我改了配置但程序读到的还是旧值”的诡异情况。排查时可以用conf.get(fs.defaultFS)打印出来确认一下不要盲改。2. HDFS FileSystem API比Shell更灵活的客户端2.1 获取FileSystem对象时绕不开的几个选择题HDFS客户端API最核心的类是抽象的FileSystem。它不是让你直接new FileSystem()而是通过工厂方法拿到具体的实现类。本地文件系统返回LocalFileSystemHDFS返回DistributedFileSystem具体取决于URI的协议头。代码里最常见的获取方式有三种看起来差不多但细节有区别Configuration conf new Configuration(); // 方式一完全依赖classpath中的配置 FileSystem fs FileSystem.get(conf); // 方式二指定HDFS URI读配置里的其它项 FileSystem fs2 FileSystem.get(URI.create(hdfs://hadoop-master:8020), conf); // 方式三指定URI、配置和操作用户 FileSystem fs3 FileSystem.get(URI.create(hdfs://hadoop-master:8020), conf, hdfs);第一种适合你已经把core-site.xml放进classpath的情况代码最简洁。第二种适合那些没有配置文件、只想临时连一个固定集群的场景。第三种则多了一个用户参数它指定了HDFS操作时的身份。这一点在本地开发时特别有用。如果你本机的系统用户名是dev但集群上没有给这个用户开放目录权限默认拿到的用户身份就会触发权限错误。显式指定一个已有权限的用户名能省去不少调试时间。用完之后记得关闭FileSystem实例最简单的方式就是放到try-with-resources里try (FileSystem fs FileSystem.get(conf)) { // 执行文件操作 }不过要注意如果你通过同一种方式连续获取多个FileSystem实例Hadoop底层可能返回同一个对象多次close()会重复关闭。所以写工具类时最好用一个全局实例程序退出时统一释放而不是每个方法里都关。2.2 上传、下载、删除和遍历的真实代码模板我最常写的一组FileSystem操作就四个上传、下载、删除、遍历。先看上传Path localSrc new Path(/home/me/logs/access.log); Path hdfsDst new Path(/data/logs/access.log); fs.copyFromLocalFile(false, true, localSrc, hdfsDst);copyFromLocalFile有多个重载参数很容易记混。前两个布尔值第一个delSrc表示复制成功后是否删除本地源文件第二个overwrite表示目标已存在时是否覆盖。我的建议是默认都传false然后在做覆盖之前先判断目标存在情况避免误删源文件。如果目标已经存在且需要更新可以先delete再复制或者允许overwrite为true业务逻辑会清晰很多。下载对应的方法就是copyToLocalFile不过我更常用的是组合open和IOUtilstry (FSDataInputStream in fs.open(new Path(/data/logs/access.log)); OutputStream out new FileOutputStream(/home/me/logs/access.log)) { IOUtils.copyBytes(in, out, 4096, true); }IOUtils.copyBytes是Hadoop自带的流复制工具最后一个布尔值true表示复制完自动关闭流。这个方法不仅适用于HDFS下载任何InputStream到OutputStream的复制场景都能用是很实用的小工具。删除文件时要注意递归参数fs.delete(new Path(/data/old), true);第二个参数true表示递归删除。只删除空目录时不需要递归但你要删除一个非空目录不传true会直接抛异常。对于生产环境的删除操作我通常会在代码里先确认路径前缀防止因为路径拼接错误导致删错目录。遍历文件和Shell的ls -R类似但API返回的是RemoteIteratorFileSystem fs FileSystem.get(conf); Path inputPath new Path(/data); RemoteIteratorLocatedFileStatus iterator fs.listFiles(inputPath, true); while (iterator.hasNext()) { LocatedFileStatus status iterator.next(); if (status.isFile()) { System.out.println(status.getPath() size status.getLen()); } }这里第二个参数true表示是否递归遍历子目录。注意listFiles只返回文件不会返回目录本身。如果你还想看目录信息可以用listStatus。遍历得到的LocatedFileStatus里除了路径和大小还有block位置、副本数、修改时间等元数据做文件统计时很好用。2.3 路径、权限和回收站那些说不完的坑HDFS API里的Path并不总是你以为的路径。比如你写new Path(/tmp/a.txt)这天然是HDFS上的绝对路径。如果你在前面加了s3a://或file:///那FileSystem就会解析成其它存储系统。很多人会把本机路径当成集群路径直接传结果发现上传时找不到文件需要明确一下copyFromLocalFile的第一个参数是客户端本地文件系统路径第二个参数才是目标文件系统路径。权限问题也是最常见的坑。错误信息通常是org.apache.hadoop.security.AccessControlException: Permission denied: userdev, accessWRITE, inode/data:root:supergroup:drwxr-xr-x这个报错说明你当前HDFS操作身份是dev但目标目录属于root且权限是755。解决思路有三个第一在FileSystem.get时指定有权限的用户像我前面写的那样第二在代码里对目录执行fs.setOwner修改所属用户第三如果是学习环境可以临时把dfs.permissions.enabled设为false重启NameNode但这只能用于本地测试生产环境绝不能关权限。还有一点容易被忽略当你调用copyFromLocalFile时如果本地文件还在写入中上传到HDFS的会是一份有截断风险的文件。所以做文件流转任务时我通常先传到一个.tmp后缀的临时路径等上传完毕确认无误后再用rename改成正式文件名这个习惯能救你很多次。3. 用MapReduce Java API跑通一个作业3.1 新版API三段式Mapper、Reducer、DriverMapReduce的Java API分为新旧两套。旧版在org.apache.hadoop.mapred包下新版在org.apache.hadoop.mapreduce包下。我默认建议学新版因为新版对作业控制更清晰而且社区里现成的大多数案例都是新版写法。核心结构是三个类Mapper、Reducer、Driver。Mapper负责处理输入的每一行Reducer负责汇总相同Key的结果Driver负责组装作业参数并提交。以经典的WordCount为例记录一下我实际跑通的代码public class WordCountMapper extends MapperLongWritable, Text, Text, IntWritable { private final static IntWritable one new IntWritable(1); private final Text word new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] words value.toString().split(\\s); for (String w : words) { word.set(w); context.write(word, one); } } }Mapper的输入Key是行偏移量输入Value是整行文本。输出Key是单词输出Value是数字1。为什么要用Text和IntWritable而不直接用Java的String和Integer因为Hadoop需要把数据序列化后在不同节点间传输。它自己的Writable机制比Java默认序列化更紧凑而且不要求完整类信息网络传输开销低得多。Reducer这边要处理Mapper形成的(单词, [1,1,1...])这样的分组public class WordCountReducer extends ReducerText, IntWritable, Text, IntWritable { Override protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { int sum 0; for (IntWritable val : values) { sum val.get(); } context.write(key, new IntWritable(sum)); } }这里有一个面试和实际开发都容易考到的点values是一个可重用的迭代对象你只能在这个reduce方法内部使用不能把它的引用存到集合里等下一轮再用。若一定要保存必须自己new出新的对象再复制内容。最后是Driver即作业入口public class WordCount { public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, word count); job.setJarByClass(WordCount.class); job.setMapperClass(WordCountMapper.class); job.setCombinerClass(WordCountReducer.class); job.setReducerClass(WordCountReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }这里的setJarByClass是很多新手的盲区。提交到集群时YARN需要在NodeManager上启动一个独立的Java进程来运行Mapper和Reducer。如果不知道去哪找这个类的jar包进程根本没法启动。setJarByClass会根据传入的Class自动定位所在的jar文件。本地模式下它不明显但提交集群时必须设置。3.2 作业参数与提交前的检查清单写Driver时我习惯在提交前先过一遍参数清单能省下很多无意义的失败重试。第一输入输出路径必须检查。输出路径是HDFS上的路径而且MapReduce要求输出路径在提交前不能存在否则直接报FileAlreadyExistsException。如果你希望重跑作业可以在Driver里先判断输出路径并删除Path outputPath new Path(args[1]); FileSystem fs outputPath.getFileSystem(conf); if (fs.exists(outputPath)) { fs.delete(outputPath, true); }第二Reducer数量不一定越多越好。虽然job.setNumReduceTasks可以显式指定但如果数据量很小设置几十个Reducer反而会造成大量小文件输出后续读取效率很低。我经验是先用默认的1个Reducer把功能跑通再根据数据量调整并行度。第三如果存在“分组后不影响最终结果”的中间操作可以加Combiner。在WordCount里求和操作满足交换律和结合律所以CombinerClass可以直接复用Reducer类。这会先在Map端做一次部分聚合减少Shuffle阶段的数据量。但要注意并非所有Reducer逻辑都能用作Combiner比如求平均值就不行。第四根据实际业务设置内存参数。如果Mapper需要装载较大字典可以在集群提交命令里通过-Dmapreduce.map.memory.mb2048调整内存也可以在代码里conf.set。默认值对数据倾斜或复杂任务常常不够用。最后是提交方式。在Hadoop 3.x下可以用hadoop jar myapp.jar com.example.WordCount /input /output如果你的jar里只有一个main类可以直接省略类名。多个main类时一定要写全限定名。yarn jar命令和hadoop jar在多数版本里等价但我更习惯用hadoop jar兼容面更广。3.3 本地跑通验证与集群提交的真实差别不要一上来就把作业提交到集群。先在本地模式下跑通逻辑速度会快很多。在本地模式下MapReduce会在一个JVM进程里用多线程模拟分布式执行不走YARN。设置方式很简单mapreduce.framework.namelocal fs.defaultFSfile:///在代码里可以把输入输出路径设为本地目录直接运行main方法。本地模式是验证业务逻辑的利器但也容易产生误导。比如你明明没设置setJarByClass本地模式照样能跑到了集群就直接报ClassNotFoundException。再比如本地模式下Reducer的并行度模拟并不严格你对Reduce数量的性能测试结果参考价值有限。等到本地跑通再真正提交到集群时我建议顺一遍这个顺序确认HDFS输入路径存在确认输出路径已被清理确认jar包里的依赖不包含集群已有的Hadoop相关包确认提交机上能通过hadoop fs -ls /访问集群。这一步如果失败大多数不是代码问题而是配置文件或版本问题。开始跑以后输出日志里会显示作业ID比如job_1720000000000_0001。如果后面需要看具体Task日志可以用yarn logs -applicationId application_xxx来拉取不要傻傻地只看Driver进程的stdout。4. 常见问题排查与避坑记录4.1 环境问题winutils、依赖冲突和一些日常报错很多用Windows开发的同学第一次跑Hadoop客户端代码时会遇到一个很烦人的报错Unable to load native-hadoop library for your platform...这个提示本身通常不影响功能但如果后面跟着权限异常比如Failed to set permissions of path那就是因为Windows下缺少Hadoop本地二进制支持导致的。解决方法是下载与Hadoop版本匹配的winutils.exe放到一个目录下然后在代码里设置System.setProperty(hadoop.home.dir, D:\\hadoop);或者直接把这个目录配置为环境变量HADOOP_HOME。不过我更推荐在Linux环境或者容器里做开发测试可以少一大类问题。不是不能解决而是这类问题属于“和业务逻辑无关的环境损耗”没必要硬扛。依赖冲突也很典型。最常遇到的是javax.xml.bind相关类找不到这通常是因为JDK9以上把一些模块移除了而Hadoop某些版本还依赖它们。处理方式是在Maven里显式加入对应的javax.xml.bind:jaxb-api依赖而不是去改Hadoop自身的依赖。另一个常见冲突是protobuf-java版本不一致这通常和Hadoop版本强相关不要随便升高版本否则NameNode之间的通信协议都会出问题。4.2 运行期ClassNotFoundException 与 MapReduce框架参数集群提交之后如果日志里出现ClassNotFoundException: com.example.WordCountMapper那大概率是Driver类所在jar没有被正确携带。检查job.setJarByClass是否设置或者提交命令里的jar路径是否错误。如果你用的是Maven打包但jar里没有包含第三方依赖而你的Mapper里又用到了第三方类那就要么把依赖打成fat jar要么确保这些第三方依赖在集群的classpath里存在。我踩过的一个比较深的坑是本地IDE运行正常但打包提交后报ClassNotFoundException。排查了好久发现是Maven的maven-jar-plugin默认打包后资源文件没有包含core-site.xml导致集群提交时客户端配置缺失。虽然‘找不到类’和‘找不到配置’看起来是两个问题但报错形式都指向运行环境不一致。所以打包之后先解压jar看看BOOT-INF/classes或者根目录下有没有你要的配置文件这是简单有效的自检步骤。另一个运行期常见问题是Container启动失败日志提示内存不足Container [pid...] is running beyond virtual memory limits这不一定是物理内存不足而是Hadoop把容器虚拟内存上限算得很保守。可以用-Dmapreduce.reduce.memory.mb和-Dmapreduce.reduce.cpu.vcores上调。如果只是一两个大任务偶尔失败给内存加上限就能过如果任务本身有内存泄漏调参数也只是延缓报错时间。4.3 问题速查表为了方便回顾我把前面提到的几个高频问题整理成了表现象常见原因处理方式连接超时或NameNode拒绝连接fs.defaultFS配置错误或机器不在同一网络检查core-site.xml用hadoop fs -ls /验证连通性Permission denied当前用户无目标目录权限在FileSystem.get指定用户或调整目录权限输出路径已存在MapReduce要求输出目录不存在在Driver中启动前删除输出目录ClassNotFoundException未设置setJarByClass或依赖没有一起打包设置setJarByClass用mvn package后检查jar内容Container内存溢出默认内存参数过小适当调大mapreduce.map.memory.mb等参数本地运行正常集群失败本地模式与集群模式环境差异检查jar内的配置文件、依赖版本、main类参数这张表是我后来做项目时反复对照的清单遇到问题先查表能解决八成的重复性故障。5. 从一个完整小工具反推Java API的组织方式5.1 需求设想做一个日志文件归档工具为了不让自己停留在“会写WordCount”的层面我做了一个模拟项目X一个日志文件归档工具。需求很简单把本地一个临时目录下每天产生的日志文件上传到HDFS上以日期命名的目录中并生成一份上传清单。这个工具正好能把FileSystem API串起来用。设计的时候我先定了几条原则配置文件统一外置路径和日期通过参数传入操作过程可重复执行上传前检查文件是否已经存在。如果已经在HDFS上有同名文件就跳过而不是覆盖防止因为误操作覆盖历史日志。5.2 关键代码实现上传的核心方法大概是这样的public class HdfsArchiveTool { private final FileSystem fs; public HdfsArchiveTool(String hdfsUri) throws IOException { Configuration conf new Configuration(); conf.set(fs.defaultFS, hdfsUri); fs FileSystem.get(conf); } public void archive(String localBaseDir, String hdfsBaseDir, String date) throws IOException { File localDir new File(localBaseDir, date); if (!localDir.isDirectory()) { return; } Path targetDir new Path(hdfsBaseDir, date); if (!fs.exists(targetDir)) { fs.mkdirs(targetDir); } File[] files localDir.listFiles(); if (files null) { return; } for (File f : files) { Path targetPath new Path(targetDir, f.getName()); if (fs.exists(targetPath)) { System.out.println(skip f.getName()); continue; } fs.copyFromLocalFile(new Path(f.getAbsolutePath()), targetPath); System.out.println(uploaded f.getName()); } } }从这段代码可以看出整个调用链其实很直白先拿到FileSystem然后用exists和mkdirs保证目标目录存在再用copyFromLocalFile上传。真正复杂的不是API本身而是围绕它做的幂等、日志和异常处理。如果你要把这个工具嵌入定时任务我建议单独起一个Java进程避免在同一个进程里长时间持有FileSystem对象后因为网络波动导致连接失效。定时框架触发时重新创建FileSystem用完再关闭保证每次任务都是干净状态。5.3 实测心得与后续扩展想法做这个小工具的时候我最大的感受是Hadoop Java API的上手成本并不高真正花费时间的全是在配置、打包和环境适配上面。比如同一个上传代码在Linux开发机上跑得很稳换到Windows上就报权限异常同一个jar在本地跑版本兼容在集群上因为依赖打包方式不对又翻车。所以我现在养成了一个习惯把环境准备和打包规则写进项目的README而不是只记在博客收藏夹里。这个工具如果继续扩展可以往两个方向走。一个是和MapReduce结合归档之后触发一个解析任务对日志做统计后输出到另一个HDFS目录另一个是接入元数据表把上传文件清单的路径、大小、文件数写到数据库里方便后续查询。你会发现一旦把HDFS的Java API和业务逻辑串起来Hadoop就从一个“需要手动操作的黑盒”变成一个可以嵌入业务系统的存储底座。最后分享一个我自己的体会学习Hadoop Java API时不用追求把所有类都记住只需要把FileSystem、Configuration、Job、Mapper、Reducer这几个核心入口用熟再配合一个可复现的打包部署流程就已经能解决大部分实际问题了。遇到不懂的类和报错先看官方Javadoc再看单元测试代码比盲目搜索靠谱得多。