利用mapreduce清洗日志

package com.libc;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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 Process {

	public static class TokenizerMapper extends
			Mapper<Object, Text, Text, Text> {
		private Text word = new Text();

		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {

			// TODO Auto-generated method stub
			String datas = "";
			try {
				datas = new String(value.getBytes(), 0, value.getLength(),
						"GBK");
			} catch (UnsupportedEncodingException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			// datas = value.toString();
			try {

				String[] split = datas.split(" time=");

				// 处理头中包含空格的字段
				Pattern p = Pattern.compile("phonemodel=\"(.*?)\"");
				String pm = getIndex(split[0], p);
				split[0] = split[0].replaceAll(pm, pm.replace(" ", ""));
				Pattern p1 = Pattern.compile("networktype=\"(.*?)\"");
				String nt = getIndex(split[0], p1);
				split[0] = split[0].replaceAll(nt, nt.replace(" ", ""));
				for (int i = 1; i < split.length; i++) {
					String[] codes = split[i].split(" ", 4);
					int headLen = split[0].split(" ").length;
					if (headLen != 20) {
						// 丢掉错误日志
						continue;
					}
					// 处理旧版本日志判别标准：|
					if (codes[2].equals("code=\"100\"")){
						if(codes[3].indexOf("contact_name")>-1){
							codes[3] = process100(codes[3]);
						}
							codes[3] = codes[3].replace(‘ ‘, ‘#‘);
						
					}else if(codes[2].equals("code=\"101\"") ){
						if(codes[3].indexOf("message_to_")>-1){
							codes[3] = process101(codes[3]);
						}
							codes[3] = codes[3].replace(‘ ‘, ‘#‘);
					}
					else if(codes[2].equals("code=\"102\"")){
						if(codes[3].indexOf("caller_n")>-1||codes[3].indexOf("caller_d")>-1){
							codes[3] = process102(codes[3]);
						}
							codes[3] = codes[3].replace(‘ ‘, ‘#‘);
						
					}else{
						codes[3] = codes[3].replace("  ", " ");
					}
						
					String collect = split[0] + " time=" + codes[0] + " "
							+ codes[1] + " " + codes[2] + " " + codes[3];
					word.set(collect);

					context.write(word, new Text(""));
				}

			} catch (Exception e) {
				// TODO Auto-generated catch block
			}
		}
	}

	public static String process100(String code) throws Exception{
		String[] codes = code.split(" ");
		HashMap<String, Contact> hs = new HashMap<String, Process.Contact>();
		Pattern p0 = Pattern.compile("_(\\d*)=");
		Pattern p1 = Pattern.compile("\"(.*)\"");
		for (int i = 0; i < codes.length; i++) {
			if (codes[i].equals(""))
				continue;
			String index = getIndex(codes[i], p0);
			if (index == null)
				continue;
			String value = getIndex(codes[i], p1);
			Contact contact = null;
			if (hs.containsKey(index)) {
				contact = hs.get(index);
			} else {
				contact = new Contact();
			}
			if (codes[i].startsWith("contact_name_")) {
				contact.contactName = value;
			} else if (codes[i].startsWith("contact_num_")) {
				contact.contactNum = value;
			}
			contact.index = index;
			hs.put(index, contact);
		}

		return printToString(hs);
	}

	public static String process101(String code) throws Exception{
		String[] codes = code.split("\"  ");
		HashMap<String, Message> hs = new HashMap<String, Process.Message>();
		Pattern p = Pattern.compile("_(\\d*)=");
		Pattern p1 = Pattern.compile("\"(.*)");
		for (int i = 0; i < codes.length; i++) {
			String index = getIndex(codes[i], p);
			String value = getIndex(codes[i], p1);
			if (index == null)
				continue;
			Message message = null;
			if (hs.containsKey(index)) {
				message = hs.get(index);
			} else {
				message = new Message();
			}
			if (codes[i].startsWith("message_time_")) {
				message.messageTime = value;
			} else if (codes[i].startsWith("message_to_")) {
				message.messageTo = value;
			}
			message.index = index;
			hs.put(index, message);
		}

		return printToString(hs);
	}

	public static String process102(String code) throws Exception{
		String[] codes = code.split("\"  ");
		HashMap<String, CallLog> hs = new HashMap<String, Process.CallLog>();
		Pattern p = Pattern.compile("_(\\d*)=");
		Pattern p1 = Pattern.compile("\"(.*)");
		for (int i = 0; i < codes.length; i++) {
			String index = getIndex(codes[i], p);
			if (index == null)
				continue;
			String value = getIndex(codes[i], p1);
			CallLog callLog = null;
			if (hs.containsKey(index)) {
				callLog = hs.get(index);
			} else {
				callLog = new CallLog();
			}
			if (codes[i].startsWith("caller_date_")) {
				callLog.callerDate = value;
			} else if (codes[i].startsWith("caller_duration_")) {
				callLog.callerDuration = value;
			} else if (codes[i].startsWith("caller_name_")) {
				callLog.callerName = value;
			} else if (codes[i].startsWith("caller_num_")) {
				callLog.callerNum = value;
			}
			callLog.index = index;
			hs.put(index, callLog);
		}

		return printToString(hs);
	}

	public static String printToString(Map hs) {
		Set set = hs.keySet();
		Iterator<String> it = set.iterator();
		String result = "";
		while (it.hasNext()) {
			result = result + hs.get(it.next()).toString() + "|";
		}
		return result;
	}

	public static String getIndex(String code, Pattern p) {
		String index = null;

		Matcher matcher = p.matcher(code);
		if (matcher.find()) {
			index = matcher.group(1);
		}
		return index;
	}

	public static class IntSumReducer extends Reducer<Text, Text, Text, Text> {

		public void reduce(Text key, Text rr, Context context)
				throws IOException, InterruptedException {
			context.write(key, new Text(""));
		}
	}

	public static class Contact {

		public String index;
		public String contactName;
		public String contactNum;

		@Override
		public String toString() {
			// TODO Auto-generated method stub
			return "contact_" + index + "=" + this.contactName + ";"
					+ this.contactNum;
		}
	}

	public static class Message {
		public String index;
		public String messageTime;
		public String messageTo;

		@Override
		public String toString() {
			// TODO Auto-generated method stub
			return "message_" + this.index + "=" + this.messageTo + ";"
					+ this.messageTime;
		}
	}

	public static class CallLog {
		public String index;
		public String callerDuration;
		public String callerNum;
		public String callerName;
		public String callerDate;

		@Override
		public String toString() {
			// TODO Auto-generated method stub
			return "callLog_" + this.index + "=" + this.callerName + ";"
					+ this.callerNum + ";" + this.callerDate + ";"
					+ this.callerDuration;
		}
	}

	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: process <in> <out>");
			System.exit(2);
		}
		Job job = new Job(conf, "process");
		job.setJarByClass(Process.class);
		job.setMapperClass(TokenizerMapper.class);
		job.setCombinerClass(IntSumReducer.class);
		job.setReducerClass(IntSumReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}