hadoop安装部署及Eclipse安装集成,这里不赘述了。
先说下业务需求吧,有个系统日志文件,记录系统的运行信息,其中包含DEBUG、INFO、WARN、ERROR四个级别的日志,现在想要看到所有WARN级别的日志不看别的级别日志,并且只看时间和级别。
那么打开Eclipse,找到DFS Loction在文件系统根目录下建一个名叫test1-in的文件夹,然后将日志文件上传到该文件夹里。
日志文件的格式如下,由于内容限制这里就不附上全部日志内容了。
[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'
新建一个Map/Reduce Project,项目名为maptest,点击finish创建项目(map/reduce项目只能在安装了hadoop的机器上编写)。
创建完成后在src下建立一个maptest的包,然后在改包下新建一个类,名叫MapTest,需要集成Configuration和实现Tool。代码如下:
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);
}
}
如果之前已经存在了test-out文件夹,会报错,所以在运行前必须得先删除这个文件夹,下面是结果
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
……