SpringBoot集成Hadoop实战：生产级WordCount服务构建

1. 项目概述：为什么一个“复杂版WordCount”值得花一整节来拆解？

在大数据初学者的自学路径里，“WordCount”几乎是绕不开的第一个程序——它像编程世界的“Hello World”，简单、直观、能快速验证环境是否跑通。但现实中的生产级Hadoop开发，从来不是写个map()和reduce()就能上线的。我带过不少刚从Java后端转过来的开发者，他们第一次在SpringBoot里集成Hadoop时，常卡在连hdfs://地址都解析不了，或者本地调试时死活读不到集群上的文件，更别说处理中文乱码、自定义序列化、多输入源合并、输出路径自动时间戳这类真实需求了。这个标题里的“复杂版”，不是为了炫技，而是把新手在真实项目中第一周就会撞上的5类典型问题，全塞进一个可运行、可调试、可扩展的WordCount里：它要能从HDFS和本地文件系统双源读取；要支持GBK/UTF-8混合编码自动识别；要能把结果按天分区写入HDFS；要能在SpringBoot启动时自动校验Hadoop配置有效性；还要把MapReduce作业的提交过程封装成REST接口，让前端点一下就能触发计算。换句话说，这不是教你怎么写MapReduce，而是教你怎么把它当成一个可维护、可监控、可嵌入现有Java生态的服务模块来用。关键词“SpringBoot集成hadoop”背后，本质是解决Java工程师对Hadoop生态的“最后一公里”信任问题——不是“能不能跑”，而是“敢不敢在生产环境里当核心组件用”。如果你正卡在本地IDE调试报ClassNotFoundException: org.apache.hadoop.fs.FileSystem，或者java.net.UnknownHostException: mycluster，又或者org.apache.hadoop.ipc.RemoteException: User: xxx does not belong to hadoop，那这节笔记就是为你写的。它不讲Hadoop原理，只讲怎么让SpringBoot和Hadoop在你的笔记本上、测试机上、甚至Docker容器里，稳稳当当地握一次手。

2. 整体设计思路与方案选型逻辑

2.1 为什么必须放弃“纯Hadoop原生写法”？

很多入门教程直接让新手写一个独立的WordCount.java，用ToolRunner.run()提交到YARN。这在学习MapReduce模型时很高效，但一旦要集成进SpringBoot，立刻暴露三个硬伤：

• 生命周期割裂：Hadoop的Configuration对象是静态初始化的，而SpringBoot的Bean是动态管理的。你无法在@PostConstruct里安全地注入Configuration，因为Configuration依赖的core-site.xml等文件路径可能还没被Spring的资源加载器扫描到；
• 异常不可控：ToolRunner.run()抛出的是Exception，不是Spring的RuntimeException子类，导致全局异常处理器（@ControllerAdvice）捕获不到，日志里只有一堆ExitCodeException，根本看不出是权限问题还是路径不存在；
• 配置耦合严重：所有Hadoop参数（如fs.defaultFS、yarn.resourcemanager.address）硬编码在XML里，没法用application.yml统一管理，更没法做profile切换（dev/test/prod）。

所以我选择完全绕过ToolRunner，改用JobAPI +Spring Boot Starter的组合。核心思路是：把Hadoop客户端当作一个“高阶数据库驱动”来用——就像你用spring-boot-starter-jdbc操作MySQL一样，用hadoop-client操作HDFS和YARN。这样所有Hadoop交互都变成Spring管理的Bean，可以加事务（虽然Hadoop本身无事务）、加AOP日志、加健康检查，甚至能用Actuator暴露Hadoop连接状态。

2.2 “复杂版”的四个分层设计原则

这个WordCount不是功能堆砌，而是按企业级服务标准分层设计的：

1. 接入层（API Layer）：提供RESTful接口POST /api/wordcount/submit，接收JSON参数（{ "inputPaths": ["/data/log1.txt", "hdfs://mycluster/data/log2.txt"], "outputPath": "/result/wordcount", "encoding": "auto" }）。这里不做任何业务逻辑，只做参数校验和任务分发，符合SpringMVC最佳实践；
2. 协调层（Orchestration Layer）：WordCountService负责组装Job对象、设置Mapper/Reducer类、配置输入输出格式。关键点在于所有Hadoop配置项都来自Spring Environment，比如spring.hadoop.fs.defaultFS=${HADOOP_FS:file:///}，这样本地开发用file://，测试环境用hdfs://test-cluster，一键切换；
3. 执行层（Execution Layer）：HadoopJobExecutor封装Job.submit()调用，但它不直接返回boolean，而是返回JobExecutionResult对象，包含jobId、startTime、durationMs、status（SUCCEEDED/FAILED/KILLED）和counters（如MAP_INPUT_RECORDS）。这个对象会被存入内存队列，供后续查询；
4. 存储层（Storage Layer）：输出路径自动添加时间戳，如/result/wordcount/20240520_143022，避免重复提交覆盖结果；同时生成_SUCCESS空文件和_LOG目录，方便运维脚本判断作业完成状态。

这种分层不是为了炫技，而是为后续扩展留出空间——比如明天要加Kafka实时流处理，只需新增一个KafkaWordCountService实现同一接口；要加权限控制，只需在接入层加@PreAuthorize("hasRole('HADOOP_ADMIN')")。

2.3 工具链选型：为什么是Hadoop 3.3.6 + Spring Boot 2.7.18？

版本选择不是拍脑袋，而是踩坑后定的“黄金组合”：

• Hadoop 3.3.6：这是Hadoop 3.x系列最后一个JDK 8兼容版本。Hadoop 3.4+强制要求JDK 11，而大量企业SpringBoot老项目还在用JDK 8。更重要的是，3.3.6修复了3.3.0里著名的FileSystem.close()并发bug——当多个线程同时关闭同一个FileSystem实例时，会抛NullPointerException，这在SpringBoot的多线程调度场景下高频出现；
• Spring Boot 2.7.18：这是2.7.x系列的最终维护版，对Hadoop 3.3.x的hadoop-common、hadoop-client依赖兼容性最好。我试过Spring Boot 3.0+，它默认使用jakarta.*命名空间，而Hadoop 3.3.x还停留在javax.*，导致Configuration类加载失败，必须手动排除jakarta.annotation-api并降级，得不偿失；
• Maven依赖策略：采用<scope>provided</scope>声明Hadoop依赖，因为Hadoop集群节点上已有这些jar包。本地开发时用maven-shade-plugin打包成fat jar，但排除所有hadoop-*和htrace-core，只保留业务代码和Spring Boot启动器。这样打出来的jar包只有2MB，而不是200MB，部署快、排查问题也快。

提示：不要用spring-hadoop-boot-starter！这个项目早在2018年就停止维护，且只支持Hadoop 2.x。它的自动配置会强行覆盖你手动设置的Configuration，导致fs.defaultFS失效，是新手最常掉的坑。

3. 核心细节解析与实操要点

3.1 Hadoop配置的“三重校验”机制

SpringBoot的application.yml里不能只写spring.hadoop.fs.defaultFS=hdfs://mycluster就完事。Hadoop客户端启动时需要至少3个核心配置才能工作，缺一不可。我的做法是建立一个HadoopConfigValidatorBean，在Spring容器启动时主动校验：

spring: hadoop: fs: defaultFS: ${HADOOP_FS:hdfs://mycluster} yarn: resourcemanager: address: ${YARN_RM_ADDRESS:mycluster:8032} core: site: location: classpath:hadoop/core-site.xml

校验逻辑分三步：

1. URL格式校验：用正则^hdfs://[a-zA-Z0-9.-]+(:[0-9]+)?$检查defaultFS是否合法。如果值是file:///，跳过后续集群校验，直接走本地模式；
2. 文件系统连通性校验：创建FileSystem实例并调用exists(new Path("/"))。这里有个关键技巧：必须设置fs.defaultFS为file:///时，FileSystem.get()才返回LocalFileSystem，否则会尝试连接HDFS NameNode。所以校验前先临时覆盖配置：

   Configuration conf = new Configuration(); conf.set("fs.defaultFS", hadoopProperties.getFs().getDefaultFS()); FileSystem fs = FileSystem.get(conf); boolean isConnected = fs.exists(new Path("/")); // 这里会触发实际网络连接

3. YARN资源管理器校验：用YarnClient连接resourcemanager.address，调用getYarnClusterMetrics()。如果超时，说明YARN没起来或网络不通。

这三重校验放在@PostConstruct里，SpringBoot启动失败时会明确报错：“Hadoop NameNode连接超时，请检查HADOOP_FS环境变量”，而不是等到用户点击提交按钮时才弹出UnknownHostException。

3.2 中文编码自动识别的实战方案

日志文件的编码从来不是统一的。线上可能有GBK的Windows日志、UTF-8的Linux日志、甚至ISO-8859-1的旧系统日志。Hadoop的TextInputFormat默认用UTF-8读取，遇到GBK文件直接乱码。我的解决方案是在Mapper端做编码探测，而不是在HDFS层转换：

• 第一步：自定义EncodingAwareLineRecordReader，继承LineRecordReader，重写initialize()方法；
• 第二步：读取文件前1024字节，用juniversalchardet库探测编码（比ICU4J轻量，jar包仅150KB）；
• 第三步：将探测结果存入Context.getConfiguration().set("input.encoding", detected)，供Mapper的setup()方法读取；
• 第四步：Mapper的map()方法里，用new String(value.getBytes(), encoding)重新解码。

关键代码片段：

public class EncodingAwareLineRecordReader extends LineRecordReader { private String detectedEncoding = "UTF-8"; @Override public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { super.initialize(split, context); // 探测逻辑：读取split的前1024字节 FileSplit fileSplit = (FileSplit) split; Path path = fileSplit.getPath(); FileSystem fs = path.getFileSystem(context.getConfiguration()); try (FSDataInputStream in = fs.open(path)) { byte[] buffer = new byte[1024]; int len = in.read(buffer); if (len > 0) { CharsetDetector detector = new CharsetDetector(); detectedEncoding = detector.detectCharset(buffer, len); } } context.getConfiguration().set("input.encoding", detectedEncoding); } }

注意：juniversalchardet对纯ASCII文本返回UTF-8，对GBK文本返回GBK，准确率98%。不要用Apache Tika，它依赖太多，会把你的SpringBoot jar包撑到50MB。

3.3 多输入源合并的路径解析策略

FileInputFormat.addInputPath()只接受单个Path，但业务需求常是“分析今天所有日志”，路径可能是/log/app1/20240520/*.log（HDFS）和/tmp/local_logs/*.log（本地）。我的做法是预处理路径列表，生成统一的List<InputSplit>：

• 对每个输入路径字符串，先判断协议头：hdfs://走DistributedFileSystem，file://或无协议走LocalFileSystem；
• 调用listStatus()获取所有匹配文件（支持glob，如*.log）；
• 将每个FileStatus转换为FileSplit，存入ArrayList；
• 最终在Job配置中，用MultipleInputs.addInputPath()逐个添加，同时指定各自的InputFormatClass。

这样做的好处是：HDFS路径和本地路径可以混用，且MultipleInputs支持为不同路径指定不同的Mapper类——比如HDFS日志用LogMapper，本地调试文件用DebugLogMapper，互不影响。

3.4 输出路径的时间戳生成与幂等性保障

FileOutputFormat.setOutputPath(job, new Path("/result/wordcount"))的问题是：重复提交会覆盖上次结果。生产环境要求“每次运行生成独立目录”，且目录名要能反映时间。我的方案是：

• 在WordCountService.submitJob()方法里，用DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss").format(LocalDateTime.now())生成时间戳；
• 拼接为/result/wordcount/20240520_143022；
• 关键一步：在调用job.waitForCompletion(true)前，先检查该路径是否存在。如果存在，抛出IllegalStateException("Output path already exists: " + outputPath)，强制用户修改参数。这比覆盖更安全，因为MapReduce作业一旦开始，中途失败可能导致部分文件残留，覆盖会丢失数据；
• 作业成功后，自动在输出目录下创建_METADATA文件，内容为JSON：{"submitTime":"2024-05-20T14:30:22", "inputPaths":["/log/app1/20240520/*.log"], "user":"admin"}，方便审计。

这个设计让WordCount具备了生产级作业的最小特征：可追溯、可重放、防误操作。

4. 实操过程与核心环节实现

4.1 环境准备：5分钟搭建本地伪分布式集群

别被“Hadoop集群”吓住。本地开发用伪分布式（Pseudo-Distributed）模式，所有进程（NameNode、DataNode、ResourceManager、NodeManager）跑在同一台机器，配置文件和真集群完全一致。步骤极简：

1. 安装Java 8：sudo apt install openjdk-8-jdk（Ubuntu）或brew install openjdk@8（Mac），确认java -version输出1.8.0_xxx；
2. 下载Hadoop 3.3.6：去官网下载tar.gz包，解压到/opt/hadoop；
3. 配置SSH免密登录（关键！）：

   ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys chmod 0600 ~/.ssh/authorized_keys # 测试：ssh localhost 应该直接登录，不输密码

4. 编辑/opt/hadoop/etc/hadoop/core-site.xml：

   <configuration> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property> </configuration>

5. 编辑/opt/hadoop/etc/hadoop/hdfs-site.xml：

   <configuration> <property> <name>dfs.replication</name> <value>1</value> </property> <property> <name>dfs.namenode.name.dir</name> <value>file:/opt/hadoop/data/namenode</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>file:/opt/hadoop/data/datanode</value> </property> </configuration>

6. 格式化NameNode：/opt/hadoop/bin/hdfs namenode -format；
7. 启动HDFS：/opt/hadoop/sbin/start-dfs.sh；
8. 验证：浏览器打开http://localhost:9870，能看到NameNode UI，点“Utilities” → “Browse the file system”，应该能看到根目录。

实操心得：如果start-dfs.sh报错JAVA_HOME is not set，编辑/opt/hadoop/etc/hadoop/hadoop-env.sh，取消注释并修改：export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64（路径用readlink -f $(which java)确认）。

4.2 SpringBoot项目结构与核心代码实现

项目采用标准Maven结构，关键目录如下：

src/main/ ├── java/com/example/wordcount/ │ ├── config/HadoopConfig.java # @Configuration类，创建FileSystem、YarnClient等Bean │ ├── controller/WordCountController.java # REST接口 │ ├── service/WordCountService.java # 业务逻辑主入口 │ ├── mapper/WordCountMapper.java # 自定义Mapper，支持编码探测 │ ├── reducer/WordCountReducer.java # 标准Reducer │ └── model/JobExecutionResult.java # 作业执行结果DTO └── resources/ ├── application.yml # SpringBoot配置 └── hadoop/ # Hadoop配置文件副本（用于本地调试） ├── core-site.xml └── hdfs-site.xml

HadoopConfig.java核心代码：

@Configuration @ConditionalOnProperty(name = "spring.hadoop.enabled", havingValue = "true", matchIfMissing = true) public class HadoopConfig { @Bean @Primary public Configuration hadoopConfiguration(@Value("${spring.hadoop.fs.defaultFS}") String defaultFS, @Value("${spring.hadoop.yarn.resourcemanager.address}") String rmAddress) { Configuration conf = new Configuration(); conf.set("fs.defaultFS", defaultFS); conf.set("yarn.resourcemanager.address", rmAddress); // 关键：禁用Hadoop的自动配置，防止与Spring冲突 conf.setBoolean("fs.automatic", false); return conf; } @Bean @ConditionalOnProperty(name = "spring.hadoop.fs.defaultFS", havingValue = "file:///") public FileSystem localFileSystem(Configuration conf) throws IOException { return FileSystem.getLocal(conf); } @Bean @ConditionalOnProperty(name = "spring.hadoop.fs.defaultFS", havingValue = "hdfs://", matchIfMissing = false) public FileSystem hdfsFileSystem(Configuration conf) throws IOException { return FileSystem.get(conf); } @Bean public YarnClient yarnClient(Configuration conf) { YarnClient client = YarnClient.createYarnClient(); client.init(conf); return client; } }

WordCountController.java接口定义：

@RestController @RequestMapping("/api/wordcount") public class WordCountController { @Autowired private WordCountService wordCountService; @PostMapping("/submit") public ResponseEntity<JobExecutionResult> submitJob(@RequestBody WordCountRequest request) { try { JobExecutionResult result = wordCountService.submitJob(request); return ResponseEntity.accepted().body(result); } catch (IllegalArgumentException e) { return ResponseEntity.badRequest().body(JobExecutionResult.failed(e.getMessage())); } catch (Exception e) { return ResponseEntity.status(500).body(JobExecutionResult.failed("Job submission failed: " + e.getMessage())); } } @GetMapping("/status/{jobId}") public ResponseEntity<JobExecutionResult> getJobStatus(@PathVariable String jobId) { JobExecutionResult result = wordCountService.getJobStatus(jobId); return ResponseEntity.ok(result); } }

WordCountRequest.java请求体：

public class WordCountRequest { @NotEmpty private List<String> inputPaths; // 支持 ["hdfs://localhost:9000/log/app1.log", "/tmp/local.log"] @NotBlank private String outputPath; // 如 "/result/wordcount" private String encoding = "auto"; // auto, UTF-8, GBK private boolean overwrite = false; // 是否允许覆盖输出路径 // getters and setters... }

4.3 Mapper与Reducer的定制化实现

标准WordCount的Mapper只做分词，但生产环境需要健壮性。我的WordCountMapper做了三件事：

• 空行过滤：if (line.trim().isEmpty()) return;
• 非中文字符清洗：用正则[^\\u4e00-\\u9fa5a-zA-Z0-9\\s]替换掉特殊符号，避免"hello-world"被切分成"hello"和"world"（连字符问题）；
• 长度截断：单词长度超过50字符的丢弃，防止单词过长OOM。

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); private String encoding = "UTF-8"; @Override protected void setup(Context context) throws IOException, InterruptedException { this.encoding = context.getConfiguration().get("input.encoding", "UTF-8"); } @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); // 编码重解码（如果encoding不是UTF-8） if (!"UTF-8".equalsIgnoreCase(encoding)) { try { line = new String(line.getBytes(StandardCharsets.UTF_8), encoding); } catch (UnsupportedEncodingException e) { // 解码失败，跳过此行 return; } } if (line.trim().isEmpty()) return; // 清洗：只保留中文、英文、数字、空格 line = line.replaceAll("[^\\u4e00-\\u9fa5a-zA-Z0-9\\s]", " "); // 分词：按空白字符分割 String[] words = line.split("\\s+"); for (String w : words) { w = w.trim(); if (w.length() > 0 && w.length() <= 50) { word.set(w); context.write(word, one); } } } }

Reducer保持简洁，但加了计数器统计：

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); @Override public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); // 计数器：统计总词数 context.getCounter("WordCount", "TOTAL_WORDS").increment(sum); } }

4.4 作业提交与状态跟踪的完整流程

WordCountService.submitJob()方法是整个流程的中枢，它串联了配置、路径解析、作业构建、提交和状态注册：

@Service public class WordCountService { @Autowired private Configuration hadoopConfiguration; @Autowired private FileSystem fileSystem; @Autowired private YarnClient yarnClient; // 内存中暂存作业状态，实际项目应换为Redis private final Map<String, JobExecutionResult> jobResults = new ConcurrentHashMap<>(); public JobExecutionResult submitJob(WordCountRequest request) throws Exception { // 1. 生成唯一jobId String jobId = "wordcount_" + System.currentTimeMillis(); // 2. 解析并验证输入路径 List<Path> inputPaths = resolveInputPaths(request.getInputPaths()); if (inputPaths.isEmpty()) { throw new IllegalArgumentException("No valid input paths found"); } // 3. 生成带时间戳的输出路径 String timestamp = DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss").format(LocalDateTime.now()); Path outputPath = new Path(request.getOutputPath() + "/" + timestamp); // 4. 幂等性检查：输出路径不能已存在 if (fileSystem.exists(outputPath) && !request.isOverwrite()) { throw new IllegalStateException("Output path already exists: " + outputPath); } // 5. 构建Job Job job = Job.getInstance(hadoopConfiguration, "WordCount-" + jobId); job.setJarByClass(WordCountApplication.class); job.setMapperClass(WordCountMapper.class); job.setCombinerClass(WordCountReducer.class); job.setReducerClass(WordCountReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 6. 设置输入输出 for (Path inputPath : inputPaths) { FileInputFormat.addInputPath(job, inputPath); } FileOutputFormat.setOutputPath(job, outputPath); // 7. 提交作业（异步，避免阻塞HTTP线程） CompletableFuture<JobExecutionResult> future = CompletableFuture.supplyAsync(() -> { try { long startTime = System.currentTimeMillis(); boolean success = job.waitForCompletion(true); // true表示打印进度 long duration = System.currentTimeMillis() - startTime; JobExecutionResult result = new JobExecutionResult(); result.setJobId(jobId); result.setStartTime(new Date(startTime)); result.setDurationMs(duration); result.setStatus(success ? "SUCCEEDED" : "FAILED"); if (success) { // 读取计数器 Counters counters = job.getCounters(); result.setTotalWords(counters.findCounter("WordCount", "TOTAL_WORDS").getValue()); } return result; } catch (Exception e) { return JobExecutionResult.failed("Job execution failed: " + e.getMessage()); } }); // 8. 注册到内存Map，供/status接口查询 JobExecutionResult pending = JobExecutionResult.pending(jobId); jobResults.put(jobId, pending); // 9. 异步完成后更新状态 future.thenAccept(result -> jobResults.put(jobId, result)); return pending; } public JobExecutionResult getJobStatus(String jobId) { return jobResults.getOrDefault(jobId, JobExecutionResult.notFound(jobId)); } private List<Path> resolveInputPaths(List<String> inputPathStrings) { List<Path> paths = new ArrayList<>(); for (String pathStr : inputPathStrings) { try { Path path = new Path(pathStr); // 检查路径是否存在（支持glob） FileStatus[] statuses = fileSystem.listStatus(path); if (statuses.length == 0) { // 可能是单个文件，直接添加 if (fileSystem.exists(path)) { paths.add(path); } } else { // glob匹配到多个文件 for (FileStatus status : statuses) { if (status.isFile()) { paths.add(status.getPath()); } } } } catch (IOException e) { // 路径无效，跳过 log.warn("Invalid input path: {}, skipping", pathStr, e); } } return paths; } }

实操心得：job.waitForCompletion(true)的true参数很重要——它会实时打印MapReduce进度（如map 100% reduce 50%），方便调试。如果设为false，你只能等作业结束才能看到日志，排查问题慢十倍。

5. 常见问题与排查技巧实录

5.1 典型问题速查表

5.2 我踩过的3个深坑与独家避坑技巧

坑1：FileSystem实例的线程安全陷阱
现象：多线程并发提交WordCount作业时，偶尔报java.lang.NullPointerException在FileSystem.close()。
原因：Hadoop的FileSystem是线程不安全的，但Spring默认的@Bean是单例。多个线程共用一个FileSystem实例，close()被多次调用。
解决方案：在HadoopConfig.java中，将FileSystemBean的作用域改为@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)，每次注入都新建实例。或者更优解：不用@Bean，在WordCountService里用FileSystem.get(conf)按需创建，用完不close()（Hadoop会自动缓存和复用）。

坑2：application.yml中$符号的YAML解析错误
现象：配置spring.hadoop.fs.defaultFS=${HADOOP_FS:hdfs://localhost:9000}，启动时报Could not resolve placeholder 'HADOOP_FS'。
原因：YAML规范中$是特殊字符，Spring Boot的PropertySourcesPlaceholderConfigurer在YAML解析阶段就报错了，根本没到占位符替换环节。
解决方案：用单引号包裹整个值：spring.hadoop.fs.defaultFS: '${HADOOP_FS:hdfs://localhost:9000}'。单引号内的内容YAML解析器原样保留，交给Spring处理。

坑3：Windows路径分隔符导致HDFS路径解析失败
现象：在Windows上开发，inputPaths传C:\logs\app.log，程序报Invalid path: C:\logs\app.log。
原因：Hadoop的Path类把\当转义符，C:\logs\app.log被解析成C:logspp.log。
解决方案：在resolveInputPaths()方法里，对每个路径字符串做预处理：pathStr.replace("\\", "/")，统一转为Unix风格斜杠。这是Windows开发者必加的兼容层。

5.3 日志调试的黄金三板斧

当作业卡在map 0%或reduce 0%时，别急着重装Hadoop。按顺序执行这三步：

1. 看客户端日志：SpringBoot控制台输出的INFO级别日志，重点找Running job:和Job job_开头的行，确认jobId是否生成；
2. 看YARN日志：访问http://localhost:8088（YARN ResourceManager UI），找到对应jobId，点“Logs”链接，查看syslog文件。这里会显示ClassNotFoundException、OutOfMemoryError等真实错误；
3. 看HDFS日志：如果syslog里有Connection refused，去http://localhost:9870（NameNode UI）看“Datanodes”页，确认DataNode是否在线。如果DataNode显示0，说明HDFS没真正启动，回退到4.1节重新执行start-dfs.sh。

最后一个小技巧：在application.yml里加logging.level.org.apache.hadoop=DEBUG，Hadoop客户端会打印所有HTTP请求（如GET http://localhost:9870/jmx?qry=Hadoop:service=NameNode,name=NameNodeInfo），帮你确认网络连通性。

6. 项目收尾与生产化建议

这个“复杂版WordCount”不是终点，而是你进入Hadoop生态的起点。我在某公司落地这个方案时，后续三个月内快速扩展出了日志实时分析、用户行为漏斗计算、离线报表生成三个模块，核心复用的就是这套SpringBoot集成框架。如果你打算把它用到生产环境，有三点必须提前规划：

第一，配置中心化：把application.yml里的Hadoop配置抽到Apollo或Nacos，不同环境（dev/test/prod）用不同namespace。fs.defaultFS在prod必须是高可用的HA地址（如hdfs://mycluster），不能写死hdfs://namenode1:9000，否则NameNode挂了整个服务就崩；

第二，作业监控体系：JobExecutionResult只是基础。下一步要集成Prometheus，暴露hadoop_job_status{job_id="xxx",status="SUCCEEDED"}指标，再配Grafana看板，监控作业成功率、平均耗时、Top10慢作业；

第三，资源隔离：在YARN上为这个SpringBoot应用单独建一个Queue，配置yarn.scheduler.capacity.root.wordcount.capacity=20，避免它抢光其他业务的CPU和内存。

我自己在实际使用中发现，最