主要介紹的是自定義OutputFormat的使用，給出的需求很簡單
對現有的日誌檔案內容進行增強。
1、從原始日誌檔案中讀取資料
2、根據日誌中的一個URL欄位到外部知識庫中獲取資訊增強到原始日誌
3、如果成功增強，則輸出到增強結果目錄；如果增強失敗，則抽取原始資料中URL欄位輸出到待爬清單目錄
流程圖

程式實現

/**
 * 從資料庫（知識庫）中載入資訊，為傳進來的map初始化
 * @author 12706
 *
 */
public class MapLoaderUtils {
    /**
     * 知識庫中載入資訊，初始化map
     * @param map
     */
 

    public static void mapInit(Map<String,String> map){
        Connection conn = null;
        Statement state = null;
        ResultSet rs = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://192.168.191.2:3306/test"
 
, "root", "123456");
            System.out.println("資料庫連線成功");
            String sql = "SELECT * FROM url_rule";
            state = conn.createStatement();
            rs = state.executeQuery(sql);
            while(rs.next()){
                //初始化map   <url,content>
                //System.out.println(rs.getString(1));
 

                //System.out.println(rs.getString(2));
                map.put(rs.getString(1), rs.getString(2));
                //System.out.println(map);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            if(conn != null){
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if(state != null){
                try {
                    state.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if(rs != null){
                try {
                    rs.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    public static void main(String[] args) {
        Map<String, String> map = new HashMap<String, String>();
        mapInit(map);
        System.out.println(map);
    }
}

/**
 * 自定義的OutoutFormat
 * 拿到maptask或者reducetask傳來的kv進行處理
 * 將不同內容輸入到不同的檔案
 *  maptask或者reduceTask在最終輸出時，先呼叫OutPutFormat的getRecordWrite方法得到
 * 一個RecordWrite然後再呼叫RecordWrite的write(k,v)方法將資料寫出
 * @author 12706
 *
 */
public class LogEnhanceOutoutFormat extends FileOutputFormat<Text, NullWritable>{

    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job)
            throws IOException, InterruptedException {
        FileSystem fs = FileSystem.get(job.getConfiguration());
        //獲取增強日誌對應的輸出流 
        FSDataOutputStream enOs = fs.create(new Path("/logenhance/enhance.data"));
        //獲取待爬資訊對應的輸出流
        FSDataOutputStream toCrawOs = fs.create(new Path("/logenhance/tocraw.data"));
        //傳入兩個檔案對應的輸出流
        LogEnhanceRecordWriter recordWriter = new LogEnhanceRecordWriter(enOs, toCrawOs);
        return recordWriter;
    }
    static class LogEnhanceRecordWriter extends RecordWriter<Text,NullWritable>{

        private FSDataOutputStream enOs = null;
        private FSDataOutputStream toCrawOs = null;

        public LogEnhanceRecordWriter(FSDataOutputStream enOs, FSDataOutputStream toCrawOs) {
            this.enOs = enOs;
            this.toCrawOs = toCrawOs;
        }

        @Override
        public void write(Text key, NullWritable value) throws IOException, InterruptedException {
            //拿到的資訊要麼www.abc.com    abcdefg或者www.abc.com    tocraw
            //根據key內容來判斷將key輸出到哪個檔案
            String info = key.toString();
            if(info.contains("tocraw")){
                //輸出到待待爬檔案
                toCrawOs.write(info.getBytes());
            }else {
                //輸出待增強日誌的檔案
                enOs.write(info.getBytes());
            }
        }

        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            //關閉流
            if(enOs != null){
                enOs.close();
            }
            if(toCrawOs != null){
                toCrawOs.close();
            }
        }

    }
}

/**
 * 主程式：讀取日誌內容，根據url去知識庫看能否查到內容資訊，如果能的話，
 * 將原來那行日誌資訊增強寫入到檔案a.txt，如果沒有查到，那麼將資訊寫到到待爬檔案b.txt
 * 至於怎麼使得不同內容輸入到不同檔案則使用自定義的OutputFormat，但主程式中要指定。
 * @author 12706
 *
 */
public class MyLogEnhance {
    static class MyLogEnhanceMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        Map<String, String> logMap = new HashMap<String, String>();
        Text k = new Text();
        //
        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
            //將資訊載入到map快取中
            MapLoaderUtils.mapInit(logMap);
        }
        //讀取日誌檔案一行文字資訊，切割後第27個為url
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            //計數器，用來記錄欄位不符合>=27的行
            Counter counter = context.getCounter("mylog", "illegal_line");
            String line = value.toString();
            String[] fields = line.split("\t");
            try {
                //有的一行不一定有27個欄位
                String url = fields[26];
                //根據url去快取中查詢是否有內容資訊
                String content = logMap.get(url);
                if(content == null){
                    //沒有內容，如果為空則只輸出url到待爬清單
                    k.set(url+"\t"+"tocraw"+"\n");
                }else {
                    //知識庫（資料庫）中有資訊，增強日誌資訊
                    k.set(line+"\t"+context+"\n");
                }
                context.write(k, NullWritable.get());

            } catch (Exception e) {
                counter.increment(1);//一次加1
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(MyLogEnhance.class);
        job.setMapperClass(MyLogEnhanceMapper.class);

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

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //使用自定義OutputFormat
        job.setOutputFormatClass(LogEnhanceOutoutFormat.class);
        //指明不需要使用reduce
        job.setNumReduceTasks(0);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //儘管我們用的自定義Outputformat,但是它是繼承自FileOutputFormat
        //在FileOutputFormat中，必須輸出一個_success檔案，所以在此還需要輸出path
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        boolean exit = job.waitForCompletion(true);
        System.exit(exit?0:1);

    }
}

測試程式
工程打包上傳到linux
將日誌檔案上傳到linux
建立資料夾/logenhance/input
將日誌檔案上傳到/logenhance/input
執行程式，檢視

[[email protected] ~]# hadoop fs -mkdir -p /logenhance/input
[[email protected] ~]# hadoop fs -put 2013072404-http-combinedBy-1373892200521-log-1 /logenhance/input
[[email protected] ~]# hadoop jar logen.jar com.scu.hadoop.t.logenhanceoutput.MyLogEnhance /logenhance/input /logenhance/output
...
 mylog
                illegal_line=1
        File Input Format Counters 
                Bytes Read=61826249
        File Output Format Counters 
                Bytes Written=68329573（這些是比以前多出來的）
                [[email protected] ~]# hadoop fs -ls /logenhance
Found 4 items
-rw-r--r--   2 root supergroup   68329573 2017-10-17 07:32 /logenhance/enhance.data
drwxr-xr-x   - root supergroup          0 2017-10-17 07:00 /logenhance/input
drwxr-xr-x   - root supergroup          0 2017-10-17 07:32 /logenhance/output
-rw-r--r--   2 root supergroup          0 2017-10-17 07:32 /logenhance/tocraw.data