Hadoop MapReduce 實戰：打造分散式 Grep 文本搜尋工具

前言

Grep 是文本搜尋的經典工具。將其移植到 Hadoop MapReduce 框架，可實現海量日誌的分散式關鍵字搜尋。本文實現一個僅含 Map 階段的 Job，輸入路徑、輸出路徑、搜尋關鍵字作為參數，高效過濾包含目標字串的行。

設計思路

• Mapper：讀取每行文本，判斷是否包含關鍵字（從 Configuration 獲取），符合則輸出 (行內容, NullWritable)。
• Reducer：不需要，設置 setNumReduceTasks(0) 直接輸出 Map 結果。
• 參數傳遞：關鍵字通過 Job Configuration 透傳給 Mapper，避免硬編碼。

完整代碼實現

package com.example.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class DistributedGrep extends Configured implements Tool {

    // Mapper: 過濾包含關鍵字的行
    public static class GrepMapper extends Mapper {
        private String pattern;

        @Override
        protected void setup(Context context) {
            // 從 Configuration 獲取搜尋關鍵字
            pattern = context.getConfiguration().get("grep.pattern");
        }

        @Override
        public void map(LongWritable offset, Text line, Context context) throws IOException, InterruptedException {
            if (line.toString().contains(pattern)) {
                context.write(line, NullWritable.get());
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        // 參數檢查:  
 
        if (args.length != 3) {
            System.err.println("用法: DistributedGrep <輸入路徑> <輸出路徑> <搜尋關鍵字>");
            return 1;
        }

        Configuration conf = getConf();
        conf.set("grep.pattern", args[2]); // 傳遞關鍵字

        Job job = Job.getInstance(conf, "Distributed Grep");
        job.setJarByClass(DistributedGrep.class);

        job.setMapperClass(GrepMapper.class);
        job.setNumReduceTasks(0); // 無 Reduce 階段

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        Path input = new Path(args[0]);
        Path output = new Path(args[1]);

        // 輸出目錄存在則刪除
        FileSystem fs = output.getFileSystem(conf);
        if (fs.exists(output)) fs.delete(output, true);

        FileInputFormat.addInputPath(job, input);
        FileOutputFormat.setOutputPath(job, output);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new Configuration(), new DistributedGrep(), args);
        System.exit(exitCode);
    }
}

編譯、打包與執行

# 1. 編譯 (需 Hadoop classpath)
javac -cp $(hadoop classpath) -d . DistributedGrep.java

# 2. 打包 JAR
jar cvf grep.jar com/example/hadoop/DistributedGrep*.class

# 3. 執行
hadoop jar grep.jar com.example.hadoop.DistributedGrep \
  /user/input/logs /user/output/grep_result "ERROR"

核心細節解析

1. ToolRunner 與 Configured

實現 Tool 介面並繼承 Configured，可透過 ToolRunner.run() 自動載入 Hadoop 配置（core-site.xml, hdfs-site.xml 等），並支援通用參數（如 -D, -conf）。

2. Configuration 參數傳遞

conf.set("grep.pattern", args[2]) 在 Driver 端設置，Mapper 端透過 context.getConfiguration().get("grep.pattern") 讀取。這是 MapReduce 跨 JVM 傳遞參數的標準方式。

3. 無 Reduce 優化

setNumReduceTasks(0) 省去 Shuffle、Sort、Reduce 開銷，Map 輸出直接寫入 HDFS，極大提升過濾類任務效能。

4. 輸出目錄處理

MapReduce 要求輸出目錄不存在，否則拋出 FileAlreadyExistsException。代碼中顯式檢查並刪除，保證 Job 可重複運行。

擴展建議

• 正則表達式支援：將 contains 改為 Pattern.matches，允許複雜模式匹配。
• 大小寫不敏感：line.toString().toLowerCase().contains(pattern.toLowerCase())。
• 計數器統計：使用 context.getCounter() 統計匹配行數、總行數。
• Combiner：若需統計關鍵字頻率，可加入 Combiner/Reducer 進行聚合。

總結

分散式 Grep 展示了 MapReduce 處理「過濾/提取」類任務的優勢：水平擴展、容錯、無需管理分散式協調。本範例代碼結構標準，可作為日誌分析、數據清洗等 ETL 任務的起點模板。