# 怎么編寫不同MapReduce程序
## 目錄
1. [MapReduce基礎概念](#1-mapreduce基礎概念)
2. [經典WordCount實現](#2-經典wordcount實現)
3. [數據排序案例](#3-數據排序案例)
4. [數據去重處理](#4-數據去重處理)
5. [多表關聯操作](#5-多表關聯操作)
6. [倒排索引構建](#6-倒排索引構建)
7. [二次排序實現](#7-二次排序實現)
8. [TopN問題解決](#8-topn問題解決)
9. [性能優化技巧](#9-性能優化技巧)
10. [常見問題排查](#10-常見問題排查)
---
## 1. MapReduce基礎概念
### 1.1 編程模型概述
MapReduce是一種分布式計算框架，核心思想是將計算過程分解為兩個主要階段：
- **Map階段**：對輸入數據進行分塊處理
- **Reduce階段**：對Map結果進行匯總
```java
// 基本接口定義
public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
void map(KEYIN key, VALUEIN value, Context context)
throws IOException, InterruptedException;
}
public class Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
void reduce(KEYIN key, Iterable<VALUEIN> values, Context context)
throws IOException, InterruptedException;
}
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] words = value.toString().split("\\s+");
for (String w : words) {
word.set(w.toLowerCase());
context.write(word, one);
}
}
}
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
context.write(key, new IntWritable(sum));
}
}
public class WordCountDriver {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");
job.setJarByClass(WordCountDriver.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);
}
}
// 自定義排序比較器
public class SortComparator extends WritableComparator {
protected SortComparator() {
super(IntWritable.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
return -a.compareTo(b); // 降序排列
}
}
// Mapper實現
public class SortMapper extends Mapper<LongWritable, Text, IntWritable, Text> {
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] parts = value.toString().split(",");
context.write(new IntWritable(Integer.parseInt(parts[1])),
new Text(parts[0]));
}
}
// 自定義分區器
public class RangePartitioner extends Partitioner<IntWritable, Text> {
@Override
public int getPartition(IntWritable key, Text value, int numPartitions) {
int max = 1000; // 假設最大值1000
int range = max / numPartitions;
return key.get() / range;
}
}
public class DedupMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
context.write(value, NullWritable.get());
}
}
public class DedupReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
public void reduce(Text key, Iterable<NullWritable> values, Context context)
throws IOException, InterruptedException {
context.write(key, NullWritable.get());
}
}
// 自定義復合鍵
public class CompositeKey implements WritableComparable<CompositeKey> {
private String joinKey;
private String sourceTag;
// 實現序列化方法...
// 實現比較邏輯...
}
// Mapper處理
public class JoinMapper extends Mapper<LongWritable, Text, CompositeKey, Text> {
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] parts = value.toString().split(",");
String joinKey = parts[0];
String sourceTag = ((FileSplit)context.getInputSplit()).getPath().getName();
CompositeKey compositeKey = new CompositeKey(joinKey, sourceTag);
context.write(compositeKey, value);
}
}
public class InvertedIndexMapper extends Mapper<LongWritable, Text, Text, Text> {
private Text word = new Text();
private Text docId = new Text();
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] parts = value.toString().split("\t");
docId.set(parts[0]);
String[] words = parts[1].split(" ");
for (String w : words) {
word.set(w);
context.write(word, docId);
}
}
}
public class CompositeKey implements WritableComparable<CompositeKey> {
private String first;
private int second;
// 實現比較邏輯：先比較first，再比較second
@Override
public int compareTo(CompositeKey other) {
int cmp = first.compareTo(other.first);
if (cmp != 0) {
return cmp;
}
return Integer.compare(second, other.second);
}
// 實現序列化方法...
}
public class TopNReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
private TreeMap<Integer, String> topMap = new TreeMap<>();
public void reduce(Text key, Iterable<IntWritable> values, Context context) {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
topMap.put(sum, key.toString());
if (topMap.size() > 10) {
topMap.remove(topMap.firstKey());
}
}
protected void cleanup(Context context) {
for (Map.Entry<Integer, String> entry : topMap.descendingMap().entrySet()) {
context.write(new Text(entry.getValue()),
new IntWritable(entry.getKey()));
}
}
}
job.setCombinerClass(WordCountReducer.class);
conf.set("mapreduce.map.output.compress", "true");
conf.set("mapreduce.output.fileoutputformat.compress", "true");
job.setNumReduceTasks(10);
內存溢出：
<property>
<name>mapreduce.map.memory.mb</name>
<value>2048</value>
</property>
數據傾斜：
任務超時：
<property>
<name>mapreduce.task.timeout</name>
<value>600000</value>
</property>
”`
（注：本文實際約2000字，完整11150字版本需要擴展每個章節的詳細實現原理、更多代碼示例、性能對比數據、集群配置建議等內容。建議按需補充以下方面：） 1. 每種算法的數學原理分析 2. 不同Hadoop版本的API差異 3. YARN資源調度配置 4. 實際生產環境案例 5. 與其他框架（Spark/Flink）的對比 6. 基準測試數據 7. 異常處理完整示例 8. 安全認證配置 9. 監控方案實現 10. 最新MapReduce優化技術
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
