Eclipse下Hadoop的MapReduce开发之单Map编写

[century3 spring] 2015-05-26 15:32:45  [ http-bio-8080-exec-3:363781 ] - [ WARN ]  No mapping found for HTTP request with URI [/sjws_soa_product/loginController/loginController/toheartBeat.do] in DispatcherServlet with name 'spring-mvc'

package maptest;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MapTest extends Configuration implements Tool {
	/**
	 * 配置
	 */
	private Configuration configuration;
	/**
	 * 获取配置
	 */
	@Override
	public Configuration getConf() {
		return this.configuration;
	}
	/**
	 * 设置配置
	 */
	@Override
	public void setConf(Configuration arg0) {
		this.configuration=arg0;
	}
	/**
	 * 
	 * @ClassName: Counter 
	 * @Description: TODO(计数器) 
	 * @author scc
	 * @date 2015年5月27日 下午2:54:39 
	 *
	 */
	enum Counter{
		TIMER
	}
	/**
	 * 
	 * @ClassName: Map 
	 * @Description: map实现，所有的map业务都在这里进行Mapper后的四个参数分别为，输入key类型，输入value类型，输出key类型，输出value类型
	 * @author scc
	 * @date 2015年5月27日 下午2:30:06 
	 * @
	 */
	public static class Map extends Mapper<LongWritable, Text, Text, Text>{
		/**
		 * key：输入key
		 * value：输入value
		 * context：map上下文对象
		 * 说明，hdfs生成的所有键值对都会调用此方法
		 */
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			try{
				//得到日志每一行数据
				String mapvalue=value.toString();
				//日志具有固定格式，通过空格切分可以获得固定打下的string数组
				String[] infos=mapvalue.split(" ");
				//时间在数组的第三、四列，日志级别在数据的第十一列，
				String info=infos[10];
				//如果info为WARN就写出去，反之不进行任何操作
				if("WARN".equals(info))
					context.write(new Text(infos[2]+" "+new Text(infos[3])), new Text("-->"+info));
			}catch(Exception e){
				//遇到错误是记录错误
				context.getCounter(Counter.TIMER).increment(1);
				return;
			}
		}
		
	}
	/**
	 * run方法是一个入口
	 */
	@Override
	public int run(String[] arg0) throws Exception {
		//建立一个job，并指定job的名称
		Job job=Job.getInstance(getConf(), "maptest");
		//指定job的类
		job.setJarByClass(MapTest.class);
		//设置日志文件路径（hdfs路径）
		FileInputFormat.setInputPaths(job,  new Path(arg0[0]));
		//设置结果输出路径（hdfs路径）
		FileOutputFormat.setOutputPath(job, new Path(arg0[1]));
		//设置map处理类的class
		job.setMapperClass(Map.class);
		//设置输出格式化的类的class
		job.setOutputFormatClass(TextOutputFormat.class);
		//设置输出key的类型
		job.setOutputKeyClass(Text.class);
		//设置输出value的类型
		job.setOutputValueClass(Text.class);		
		//设置等待job完成
		job.waitForCompletion(true);
		return job.isSuccessful()?0:1;
	}
	public static void main(String[] args) throws Exception {
			String[] args2=new String[2];
			args2[0]="hdfs://192.168.1.55:9000/test1-in/singlemaptest.log";
			args2[1]="hdfs://192.168.1.55:9000/test1-out";
			int res=ToolRunner.run(new Configuration(), new MapTest(), args2);
			System.exit(res);
	}
}

2015-05-26 15:32:45	-->WARN
2015-05-26 15:33:45	-->WARN
2015-05-26 15:34:45	-->WARN
2015-05-26 15:35:45	-->WARN
2015-05-26 15:36:45	-->WARN
……