IT博客汇 | [原]日志处理实验之MapReduce方法

[原]日志处理实验之MapReduce方法

book_mmicky发表于 2014-05-13 15:23:30

1：创建日志格式处理类KPI

package hadoop2.logs;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Locale;
import java.util.Set;

/*
 * KPI Object
 */
public class KPI {
	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;// 判断数据是否合法

	private static KPI parser(String line) {
		// System.out.println(line);
		KPI kpi = new KPI();
		String[] arr = line.split(" ");
		if (arr.length > 11) {
			kpi.setRemote_addr(arr[0]);
			kpi.setRemote_user(arr[1]);
			kpi.setTime_local(arr[3].substring(1));
			kpi.setRequest(arr[6]);
			kpi.setStatus(arr[8]);
			kpi.setBody_bytes_sent(arr[9]);
			kpi.setHttp_referer(arr[10]);

			if (arr.length > 12) {
				kpi.setHttp_user_agent(arr[11] + " " + arr[12]);
			} else {
				kpi.setHttp_user_agent(arr[11]);
			}

			if (Integer.parseInt(kpi.getStatus()) >= 400) {// 大于400，HTTP错误
				kpi.setValid(false);
			}
		} else {
			kpi.setValid(false);
		}
		return kpi;
	}

	/**
	 * 按page的pv分类
	 */
	public static KPI filterPVs(String line) {
		return parser(line);
		// KPI kpi = parser(line);
		// Set<String> pages = new HashSet<String>();
		// pages.add("/about");
		// pages.add("/black-ip-list/");
		// pages.add("/cassandra-clustor/");
		// pages.add("/finance-rhive-repurchase/");
		// pages.add("/hadoop-family-roadmap/");
		// pages.add("/hadoop-hive-intro/");
		// pages.add("/hadoop-zookeeper-intro/");
		// pages.add("/hadoop-mahout-roadmap/");
		//
		// if (!pages.contains(kpi.getRequest())) {
		// kpi.setValid(false);
		// }
		// return kpi;
	}

	/**
	 * 按page的独立ip分类
	 */
	public static KPI filterIPs(String line) {
		return parser(line);
		// KPI kpi = parser(line);
		// Set<String> pages = new HashSet<String>();
		// pages.add("/about");
		// pages.add("/black-ip-list/");
		// pages.add("/cassandra-clustor/");
		// pages.add("/finance-rhive-repurchase/");
		// pages.add("/hadoop-family-roadmap/");
		// pages.add("/hadoop-hive-intro/");
		// pages.add("/hadoop-zookeeper-intro/");
		// pages.add("/hadoop-mahout-roadmap/");
		//
		// if (!pages.contains(kpi.getRequest())) {
		// kpi.setValid(false);
		// }
		//
		// return kpi;
	}

	/**
	 * PV按浏览器分类
	 */
	public static KPI filterBroswer(String line) {
		return parser(line);
	}

	/**
	 * PV按小时分类
	 */
	public static KPI filterTime(String line) {
		return parser(line);
	}

	/**
	 * PV按访问域名分类
	 */
	public static KPI filterDomain(String line) {
		return parser(line);
	}

	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append("valid:" + this.valid);
		sb.append("\nremote_addr:" + this.remote_addr);
		sb.append("\nremote_user:" + this.remote_user);
		sb.append("\ntime_local:" + this.time_local);
		sb.append("\nrequest:" + this.request);
		sb.append("\nstatus:" + this.status);
		sb.append("\nbody_bytes_sent:" + this.body_bytes_sent);
		sb.append("\nhttp_referer:" + this.http_referer);
		sb.append("\nhttp_user_agent:" + this.http_user_agent);
		return sb.toString();
	}

	public String getRemote_addr() {
		return remote_addr;
	}

	public void setRemote_addr(String remote_addr) {
		this.remote_addr = remote_addr;
	}

	public String getRemote_user() {
		return remote_user;
	}

	public void setRemote_user(String remote_user) {
		this.remote_user = remote_user;
	}

	public String getTime_local() {
		return time_local;
	}

	public Date getTime_local_Date() throws ParseException {
		SimpleDateFormat df = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss",
				Locale.US);
		return df.parse(this.time_local);
	}

	public String getTime_local_Date_hour() throws ParseException {
		SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHH");
		return df.format(this.getTime_local_Date());
	}

	public void setTime_local(String time_local) {
		this.time_local = time_local;
	}

	public String getRequest() {
		return request;
	}

	public String getRequest_domain() {
		// String str = this.request.replace("\"", "").replace("http://", "")
		// .replace("https://", "");
		// return str.lastIndexOf("/") > 0 ? str.substring(0, str.lastIndexOf("/")) : "/";
		String rtnString="";		
		String[] request_domain = request.split("/");

		if (request_domain.length > 3) {
			for (int i = 0; i < 3; i++) {
				rtnString= rtnString + request_domain[i]+"/" ;
			}
		} else {
			for (int i = 0; i < request_domain.length; i++) {
				rtnString= request.lastIndexOf("/") > 0 ? request.substring(0, request.lastIndexOf("/")) +"/" : "/";
			}
		}
		
		return rtnString;
		// string value = "192.168.128.33";
		// string[] names = value.split("\\.");
		// for (int i = 0; i < names.length; i++) {
		// system.out.println(names[i]);

	}

	public void setRequest(String request) {
		this.request = request;
	}

	public String getStatus() {
		return status;
	}

	public void setStatus(String status) {
		this.status = status;
	}

	public String getBody_bytes_sent() {
		return body_bytes_sent;
	}

	public void setBody_bytes_sent(String body_bytes_sent) {
		this.body_bytes_sent = body_bytes_sent;
	}

	public String getHttp_referer() {
		return http_referer;
	}

	public String getHttp_referer_domain() {
		if (http_referer.length() < 8) {
			return http_referer;
		}

		String str = this.http_referer.replace("\"", "").replace("http://", "")
				.replace("https://", "");
		return str.indexOf("/") > 0 ? str.substring(0, str.indexOf("/")) : str;
	}

	public void setHttp_referer(String http_referer) {
		this.http_referer = http_referer;
	}

	public String getHttp_user_agent() {
		return http_user_agent;
	}

	public void setHttp_user_agent(String http_user_agent) {
		this.http_user_agent = http_user_agent;
	}

	public boolean isValid() {
		return valid;
	}

	public void setValid(boolean valid) {
		this.valid = valid;
	}

	public static void main(String args[]) {
		String line = "222.68.172.190 - - [18/Sep/2013:06:49:57 +0000] \"GET /stru.zip HTTP/1.1\" 200 19939 \"http://www.angularjs.cn/A00n\" \"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/29.0.1547.66 Safari/537.36\"";
		System.out.println(line);
		KPI kpi = new KPI();
		String[] arr = line.split(" ");

		kpi.setRemote_addr(arr[0]);
		kpi.setRemote_user(arr[1]);
		kpi.setTime_local(arr[3].substring(1));
		kpi.setRequest(arr[6]);
		kpi.setStatus(arr[8]);
		kpi.setBody_bytes_sent(arr[9]);
		kpi.setHttp_referer(arr[10]);
		kpi.setHttp_user_agent(arr[11] + " " + arr[12]);
		// System.out.println(kpi);
		 System.out.println(kpi.getRequest_domain());

		try {
			SimpleDateFormat df = new SimpleDateFormat("yyyy.MM.dd:HH:mm:ss",
					Locale.US);
			System.out.println(df.format(kpi.getTime_local_Date()));
			System.out.println(kpi.getTime_local_Date_hour());
			System.out.println(kpi.getHttp_referer_domain());
		} catch (ParseException e) {
			e.printStackTrace();
		}
	}

}

2：统计(所有日志)独立 ip 数目，即不同 ip 的总数

下面代码输出了独立IP的访问次数，同时通过一个reduce计数器ReportTest.TotalIP来得到不同IP的总数。由于例程中combine和reduce采用同一方法，所以在统计IP 的总数的时候，需要取消combine过程，不然计数器将是combine过程计数和reduce过程计数过程之和。当然如果combine和reduce用不同的方法，那就不必取消combine过程了。

package hadoop2.logs;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class KPIIP {
	
	static enum ReportTest{
		TotalIP
	}
	public static class KPIIPMapper extends
			Mapper<Object, Text, Text, IntWritable> {
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		@Override
		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			KPI kpi = KPI.filterPVs(value.toString());
			word.set(kpi.getRemote_addr());
			context.write(word, one);
		}
	}

	public static class KPIIPReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

		@Override
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.getCounter(ReportTest.TotalIP).increment(1);
			context.write(key, result);
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: KPIIP <in> <out>");
			System.exit(2);
		}

		Job job = new Job(conf, "KPIIP");
		job.setJarByClass(KPIIP.class);
		job.setMapperClass(KPIIPMapper.class);
		//job.setCombinerClass(KPIIPReducer.class);
		job.setReducerClass(KPIIPReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}

3：统计(所有日志)每个子目录访问次数

对于日志来说，客户端的访问还有很多种形式，有可能是点击的、有可能是链接等等，日志中记录下来的访问请求要根据需要进行处理来满足分析的需求，可以通过编写KPI.filterPVs方法来达到过滤的目的，用KPI.getRequest_domain来格式化要分析的用户请求数据，最终通过mapreduce来完成分析之目的。

package hadoop2.logs;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class KPIPV {

	public static class KPIPVMapper extends
			Mapper<Object, Text, Text, IntWritable> {
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		@Override
		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			KPI kpi = KPI.filterPVs(value.toString());
			word.set(kpi.getRequest_domain());
			context.write(word, one);
		}
	}

	public static class KPIPVReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

		@Override
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(key, result);
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: KPIPV <in> <out>");
			System.exit(2);
		}

		Job job = new Job(conf, "KPIPV");
		job.setJarByClass(KPIPV.class);
		job.setMapperClass(KPIPVMapper.class);
		job.setCombinerClass(KPIPVReducer.class);
		job.setReducerClass(KPIPVReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}

4：统计(所有日志)每个 ip，访问的子目录次数

和上面的例子一样，只是map的key中多了一个IP信息。

package hadoop2.logs;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class KPI3 {
	public static class KPI3Mapper extends
			Mapper<Object, Text, Text, IntWritable> {
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		@Override
		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			KPI kpi = KPI.filterPVs(value.toString());
			word.set(kpi.getRemote_addr() + " " + kpi.getRequest_domain());
			context.write(word, one);
		}
	}

	public static class KPI3Reducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

		@Override
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(key, result);
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: KPI3 <in> <out>");
			System.exit(2);
		}

		Job job = new Job(conf, "KPI3");
		job.setJarByClass(KPI3.class);
		job.setMapperClass(KPI3Mapper.class);
		job.setCombinerClass(KPI3Reducer.class);
		job.setReducerClass(KPI3Reducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}