Hadoop MapReduce 利用 MultipleOutputs 控制 Reduce 輸出路徑

Hadoop 舊版 API (mapred 套件) 中，MultipleOutputs 類別允許 Reduce 階段將資料依邏輯分類寫入不同輸出檔案，而非單一 part-r-00000。本文示範完整實作流程。

核心程式碼結構

public class WordCount {
    public static class MyMap implements Mapper<LongWritable, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        @Override
        public void map(LongWritable key, Text value, 
                       OutputCollector<Text, IntWritable> output, Reporter reporter) 
                       throws IOException {
            StringTokenizer tokenizer = new StringTokenizer(value.toString());
            while (tokenizer.hasMoreTokens()) {
                word.set(tokenizer.nextToken());
                output.collect(word, one);
            }
        }
    }

    public static class MyReduce implements Reducer<Text, IntWritable, Text, IntWritable> {
        private MultipleOutputs mos;

        @Override
        public void configure(JobConf job) {
            mos = new MultipleOutputs(job);
        }

        @Override
        public void reduce(Text key, Iterator<IntWritable> values,
                          OutputCollector<Text, IntWritable> output, Reporter reporter)
                          throws IOException {
            int sum = 0;
            while (values.hasNext()) sum += values.next().get();

            // 依關鍵字首字母分流輸出
            String prefix = key.toString().substring(0, 1).toLowerCase();
            mos.getCollector(prefix, reporter).collect(key, new IntWritable(sum));
        }

        @Override
        public void close() throws IOException {
            mos.close();  // 必須關閉以確保資料刷新
        }
    }
}

輸出結果範例

/chukwa/output/0419-10/
├── _SUCCESS
├── _logs/
├── part-00000.lzo          # 預設輸出
├── test1-m-00000.lzo       # MultipleOutputs 產生
├── test2-m-00000.lzo
├── test3-m-00000.lzo
├── test4-m-00001.lzo
└── test5-m-00001.lzo

關鍵配置要點

1. Reduce configure() 中初始化 MultipleOutputs
2. 使用 mos.getCollector(namedOutput, reporter) 取得具名輸出收集器
3. close() 中呼叫 mos.close() 確保所有緩衝資料寫入
4. Job 配置需啟用：job.setOutputFormat(TextOutputFormat.class)

適用場景

• 多表關聯查詢結果分表存儲
• 日誌按類型/日期/等級分檔
• 資料清洗產生正常/異常資料分流
• 機器學習特徵工程輸出多種格式

遷移提示

新版 mapreduce API (org.apache.hadoop.mapreduce) 使用 MultipleOutputs 配合 Context.write(key, value, name)，語法更簡潔，建議新專案採用新版 API。