1. 从WordCount入门MapReduce

第一次接触Hadoop时，WordCount就像编程界的"Hello World"一样经典。记得我刚开始学习时，对着这个简单的词频统计案例反复琢磨了好几天。现在回头看，它确实是理解MapReduce思想的最佳切入点。

WordCount的核心逻辑非常简单：统计文本中每个单词出现的次数。但就是这样一个简单的需求，在分布式环境下需要考虑很多问题。比如，如何把大文件拆分成小块并行处理？如何汇总来自不同节点的统计结果？这些正是MapReduce要解决的核心问题。

来看一个实际的Java实现代码：

public class WordCount {
  public static class TokenizerMapper 
       extends Mapper<Object, Text, Text, IntWritable>{
       
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }
  
  public static class IntSumReducer 
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    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);
    }
  }
}

这段代码展示了MapReduce的典型结构。Mapper负责将输入文本拆分成单词并输出键值对（单词，1），Reducer则负责汇总相同单词的计数。我在第一次运行时犯了个错误，忘记设置Combiner，导致网络传输数据量过大。后来加上job.setCombinerClass(IntSumReducer.class)后性能提升了近40%。

2. HDFS文件操作基础

在真实项目中，我们首先需要把数据放到HDFS上。记得我第一次操作HDFS时，把本地文件上传后却怎么也找不到，后来才发现是路径写错了。HDFS的操作和Linux类似，但有些细节需要注意。

下面是一个完整的Java示例，演示如何创建、写入和读取HDFS文件：

public class HDFSExample {
  public static void main(String[] args) throws IOException {
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(conf);
    
    // 创建文件并写入数据
    Path file = new Path("/user/hadoop/sample.txt");
    if (!fs.exists(file)) {
      FSDataOutputStream out = fs.create(file);
      out.writeUTF("Hello HDFS");
      out.close();
    }
    
    // 读取文件内容
    FSDataInputStream in = fs.open(file);
    String content = in.readUTF();
    System.out.println("文件内容: " + content);
    
    // 获取文件元信息
    FileStatus status = fs.getFileStatus(file);
    System.out.println("文件大小: " + status.getLen());
    System.out.println("最后修改时间: " + new Date(status.getModificationTime()));
    
    fs.close();
  }
}

在实际项目中，我经常遇到文件权限问题。HDFS的权限模型和Linux类似，但默认配置下权限检查是关闭的。如果遇到Permission denied错误，可以检查以下两点：一是确保路径正确，二是检查hdfs-site.xml中的dfs.permissions配置。

3. 构建倒排索引系统

倒排索引是搜索引擎的核心技术之一，也是MapReduce的经典应用场景。与WordCount不同，倒排索引需要记录单词出现在哪些文档中，以及出现的频率。

我曾经为一个文档检索系统实现过倒排索引，核心思路是：Mapper输出格式为(单词, 文档ID@1)，Reducer将相同单词的文档ID合并。这里有个优化点：可以在Mapper端先做局部合并，减少网络传输。

完整实现代码如下：

public class InvertedIndex {
  public static class InvertedIndexMapper 
       extends Mapper<LongWritable, Text, Text, Text> {
       
    public void map(LongWritable key, Text value, Context context
                    ) throws IOException, InterruptedException {
      FileSplit split = (FileSplit)context.getInputSplit();
      String fileName = split.getPath().getName();
      
      // 使用哈希表在Mapper端做局部聚合
      Map<String, Integer> wordCounts = new HashMap<>();
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        String word = itr.nextToken();
        wordCounts.put(word, wordCounts.getOrDefault(word, 0) + 1);
      }
      
      // 输出局部聚合结果
      for (Map.Entry<String, Integer> entry : wordCounts.entrySet()) {
        context.write(new Text(entry.getKey()), 
                     new Text(fileName + "@" + entry.getValue()));
      }
    }
  }
  
  public static class InvertedIndexReducer 
       extends Reducer<Text, Text, Text, Text> {
       
    public void reduce(Text key, Iterable<Text> values, 
                       Context context
                       ) throws IOException, InterruptedException {
      StringBuilder result = new StringBuilder();
      for (Text val : values) {
        if (result.length() > 0) {
          result.append(";");
        }
        result.append(val.toString());
      }
      context.write(key, new Text(result.toString()));
    }
  }
}

在实际部署时，我发现当文档数量很大时，Reducer的内存可能不够用。这时可以考虑以下优化：1) 增加Reducer数量；2) 对输出进行压缩；3) 使用二次排序技术减少内存使用。

4. PageRank算法实现

PageRank是Google创始人提出的网页排序算法，也是MapReduce的高级应用。它通过网页之间的链接关系计算每个页面的重要性。我曾在学术研究中实现过这个算法，迭代计算过程非常适合MapReduce。

PageRank的核心公式是： PR(A) = (1-d)/N + d * Σ(PR(Ti)/C(Ti)) 其中d是阻尼系数，通常取0.85；N是网页总数；C(Ti)是页面Ti的出链数量。

Java实现的关键部分如下：

public class PageRank {
  public static class PageRankMapper 
       extends Mapper<Object, Text, Text, Text> {
       
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      String[] parts = value.toString().split("\t");
      String page = parts[0];
      double currentRank = Double.parseDouble(parts[1]);
      String[] links = Arrays.copyOfRange(parts, 2, parts.length);
      
      // 输出页面关系结构
      context.write(new Text(page), new Text("_" + String.join(" ", links)));
      
      // 计算并输出PR值贡献
      if (links.length > 0) {
        double contribution = currentRank / links.length;
        for (String link : links) {
          context.write(new Text(link), new Text(String.valueOf(contribution)));
        }
      }
    }
  }
  
  public static class PageRankReducer 
       extends Reducer<Text, Text, Text, Text> {
       
    private static final double DAMPING = 0.85;
    
    public void reduce(Text key, Iterable<Text> values, 
                       Context context
                       ) throws IOException, InterruptedException {
      double sum = 0;
      String links = "";
      
      for (Text val : values) {
        String str = val.toString();
        if (str.startsWith("_")) {
          links = str.substring(1);
        } else {
          sum += Double.parseDouble(str);
        }
      }
      
      double newRank = (1 - DAMPING) / TOTAL_PAGES + DAMPING * sum;
      String output = String.format("%.3f %s", newRank, links);
      context.write(key, new Text(output));
    }
  }
}

在真实应用中，PageRank需要多次迭代才能收敛。我通常设置10-20次迭代，每次迭代都是一个完整的MapReduce作业。为了优化性能，可以使用ChainMapper/Reducer将多次迭代合并为一个作业，避免中间结果的IO开销。

5. 性能优化实战技巧

经过多个项目的实践，我总结了一些MapReduce性能优化的经验：

1. 合理设置Reducer数量：经验法则是Reducer数量略小于集群节点数×每个节点最大容器数。可以通过mapreduce.job.reduces参数设置。

2. 使用Combiner：像WordCount这样满足结合律和交换律的操作，使用Combiner能显著减少网络传输。但要注意Combiner的输入输出类型必须与Reducer一致。

3. 数据倾斜处理：当某些键的数据量特别大时，会导致个别Reducer负载过重。解决方法包括：

   • 自定义Partitioner分散热点
   • 增加Reducer数量
   • 对热点键进行特殊处理

4. 小文件合并：HDFS不适合存储大量小文件。可以通过以下方式优化：

   // 设置输入合并
   job.setInputFormatClass(CombineTextInputFormat.class);
   CombineTextInputFormat.setMaxInputSplitSize(job, 128*1024*1024); // 128MB
   

5. 内存调优：对于内存密集型任务，需要调整JVM参数：

   <property>
     <name>mapreduce.map.memory.mb</name>
     <value>2048</value>
   </property>
   <property>
     <name>mapreduce.reduce.memory.mb</name>
     <value>4096</value>
   </property>
   

6. 压缩中间结果：可以显著减少磁盘IO和网络传输：

   conf.set("mapreduce.map.output.compress", "true");
   conf.set("mapreduce.map.output.compress.codec", 
           "org.apache.hadoop.io.compress.SnappyCodec");
   

6. 常见问题排查

在开发MapReduce程序时，经常会遇到各种问题。以下是我总结的一些常见问题及解决方法：

1. 任务卡住不执行：

   • 检查ResourceManager和NodeManager日志
   • 确认集群资源是否充足
   • 检查是否有过多任务排队

2. Reducer进度一直为0%：

   • 可能是Mapper没有输出
   • 检查Mapper是否正确调用了context.write()
   • 确认输出键值类型与Reducer输入类型匹配

3. OOM内存溢出：

   • 增加任务内存配置
   • 检查是否有数据倾斜
   • 优化代码中的集合使用，避免保存过多数据在内存中

4. HDFS权限问题：

   • 检查程序运行用户是否有权限访问输入输出路径
   • 确认HDFS的权限检查是否开启
   • 使用hdfs dfs -ls命令验证路径

5. 性能瓶颈分析：

   • 使用JobHistory Server分析任务时间分布
   • 检查是否有数据倾斜（某些Reducer处理数据量远大于其他）
   • 确认是否启用了压缩

记得有一次，我遇到一个任务运行特别慢的问题，最后发现是因为在Reducer中错误地使用了String拼接，导致产生了大量临时对象。改为使用StringBuilder后性能提升了3倍。这也提醒我们，在MapReduce中同样需要注意基本的Java编码优化。