一、 簡單解析版

1.需求

去除日誌中欄位長度小於等於11的日誌。

2.輸入資料

3.實現程式碼

（1）編寫LogMapper

package com.bigdata.mapreduce.weblog;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class LogMapper extends Mapper<LongWritable, Text, Text, NullWritable>{	
	Text k = new Text();
	
	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		
		// 1 獲取1行資料
		String line = value.toString();
		
		// 2 解析日誌
		boolean result = parseLog(line,context);
		
		// 3 日誌不合法退出
		if (!result) {
			return;
		}
		
		// 4 設定key
		k.set(line);
		
		// 5 寫出資料
		context.write(k, NullWritable.get());
   	}

	// 2 解析日誌
	private boolean parseLog(String line, Context context) {
		// 1 擷取
		String[] fields = line.split(" ");
		
		// 2 日誌長度大於11的為合法
		if (fields.length > 11) {
			// 系統計數器
			context.getCounter("map", "true").increment(1);
			return true;
		}else {
			context.getCounter("map", "false").increment(1);
			return false;
		}
	}
}
 

（2）編寫LogDriver

package com.bigdata.mapreduce.weblog;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class LogDriver {

	public static void main(String[] args) throws Exception {

      		args = new String[] { "e:/input/inputlog", "e:/output1" };

		// 1 獲取job資訊
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		// 2 載入jar包
		job.setJarByClass(LogDriver.class);

		// 3 關聯map
		job.setMapperClass(LogMapper.class);

		// 4 設定最終輸出型別
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		// 設定reducetask個數為0
		job.setNumReduceTasks(0);

		// 5 設定輸入和輸出路徑
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		// 6 提交
		job.waitForCompletion(true);
	}
}
 

二、複雜解析版

1．需求

對web訪問日誌中的各欄位識別切分
去除日誌中不合法的記錄
根據統計需求，生成各類訪問請求過濾資料

2．輸入資料

3．實現程式碼

（1）定義一個bean，用來記錄日誌資料中的各資料欄位

package com.bigdata.mapreduce.log;

@Data
public class LogBean {
	private String remote_addr;// 記錄客戶端的ip地址
	private String remote_user;// 記錄客戶端使用者名稱稱,忽略屬性"-"
	private String time_local;// 記錄訪問時間與時區
	private String request;// 記錄請求的url與http協議
	private String status;// 記錄請求狀態；成功是200
	private String body_bytes_sent;// 記錄傳送給客戶端檔案主體內容大小
	private String http_referer;// 用來記錄從那個頁面連結訪問過來的
	private String http_user_agent;// 記錄客戶瀏覽器的相關資訊
	private boolean valid = true;// 判斷資料是否合法

	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append(this.valid);
		sb.append("\001").append(this.remote_addr);
		sb.append("\001").append(this.remote_user);
		sb.append("\001").append(this.time_local);
		sb.append("\001").append(this.request);
		sb.append("\001").append(this.status);
		sb.append("\001").append(this.body_bytes_sent);
		sb.append("\001").append(this.http_referer);
		sb.append("\001").append(this.http_user_agent);
		
		return sb.toString();
	}
}
 

（2）編寫LogMapper程式

package com.bigdata.mapreduce.log;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class LogMapper extends Mapper<LongWritable, Text, Text, NullWritable>{
	Text k = new Text();

@Override
protected void map(LongWritable key, Text value, Context context)
		throws IOException, InterruptedException {
	// 1 獲取1行
	String line = value.toString();
	
	// 2 解析日誌是否合法
	LogBean bean = pressLog(line);
	
	if (!bean.isValid()) {
		return;
	}
	
	k.set(bean.toString());
	
	// 3 輸出
	context.write(k, NullWritable.get());
}

// 解析日誌
private LogBean pressLog(String line) {
	LogBean logBean = new LogBean();
	
	// 1 擷取
	String[] fields = line.split(" ");
	
	if (fields.length > 11) {
		// 2封裝資料
		logBean.setRemote_addr(fields[0]);
		logBean.setRemote_user(fields[1]);
		logBean.setTime_local(fields[3].substring(1));
		logBean.setRequest(fields[6]);
		logBean.setStatus(fields[8]);
		logBean.setBody_bytes_sent(fields[9]);
		logBean.setHttp_referer(fields[10]);
		
		if (fields.length > 12) {
			logBean.setHttp_user_agent(fields[11] + " "+ fields[12]);
		}else {
			logBean.setHttp_user_agent(fields[11]);
		}
		
		// 大於400，HTTP錯誤
		if (Integer.parseInt(logBean.getStatus()) >= 400) {
			logBean.setValid(false);
		}
	}else {
		logBean.setValid(false);
	}
	
	return logBean;
}
}

（3）編寫LogDriver程式

package com.bigdata.mapreduce.log;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class LogDriver {
	public static void main(String[] args) throws Exception {
		// 1 獲取job資訊
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

	// 2 載入jar包
	job.setJarByClass(LogDriver.class);

	// 3 關聯map
	job.setMapperClass(LogMapper.class);

	// 4 設定最終輸出型別
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(NullWritable.class);

	// 5 設定輸入和輸出路徑
	FileInputFormat.setInputPaths(job, new Path(args[0]));
	FileOutputFormat.setOutputPath(job, new Path(args[1]));

	// 6 提交
	job.waitForCompletion(true);
}
}