Web开发学习10 hadoop实战

撒的阿坎努斯

前一篇说到项目记录了很多埋点日志,当有一天项目需求需要对这些日志做分析时那hadoop就是一把好手了,下面简单介绍下用java调用hadoop分布式计算的例子
首先我们需要做hadoop配置

public static Configuration getConf(){
Configuration conf = new Configuration();
conf.set("fs.default.name", "hdfs://192.168.217.129:9100");
conf.set("mapred.job.tracker", "192.168.217.129:9101");
return conf;
}

这里面的配置项是根据服务端hadoop的配置来定的,总之要能连接的上hadoop
然后提供获取hdsf文件系统的方法

public static FileSystem getHdfs(){
if(hdfs != null){
return hdfs;
}
try {
hdfs = FileSystem.get(getConf());
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return hdfs;
}

FileSysem是hdfs的核心类,封装了很多文件处理的方法比如我们要把我们的日志文件上传到hdfs

getHdfs().copyFromLocalFile(new Path("D://op.log"), new Path(getHdfs().getWorkingDirectory().toString() + "/input/op.log"));

执行完成后在eclipse hadoop视图中就能看到对应的文件

然后我们来看下文件的内容,里面都是用户的操作记录,我们要做的任务是把操作日志中用户发送消息的消息内容提取出来,并且统计发送的次数

都知道hadoop是用mapreduce的方式来计算的,那就先来看看这两块代码

/**
* 映射器
* 用于将我们的数据进行预处理
*/
public static class MyMapper extends Mapper<LongWritable, Text, Text, Text>{
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//获取单行数据
String str = value.toString();
String [] arr = str.split("\\|");
if("/account!sendMessage.action".equals(arr[5]))
//把用户名作为key 操作数据作为值输出
context.write(new Text(arr[1]), new Text(","+arr[6]));
}
}
/**
* 处理器
* 用于将mapper预处理的数据记录进行业务计算，然后输出
*/
public static class MyReducer extends Reducer<Text, Text, Text, Text>{
@Override
//每个key都会调用这个reduce方法 values参数是用户的操作数据的集合,
//hadoop会自动把相同key的values用集合的方式存储,一起穿个reduce处理
protected void reduce(Text key, Iterable<Text> values,
Context context)
throws IOException, InterruptedException {
int i = 0;
for (Text v : values) {
//统计用户发送的次数
i ++;
context.write(new Text(key), new Text(v));
}
context.write(new Text(key), new Text(i+"次发信息"));
}
}


代码还是非常简洁的,mapper只做数据的解析,把日志数据进行拆分,索取需要的数据,reducer则做计算操作,最后输出我们想要的结果,我们要做的就是告诉hadoop输入输出的参数类型
然后再来看看如果执行这段mapreduce代码

public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
System.setProperty("hadoop.home.dir", "D:/development/hadoop-2.2.0");

//getHdfs().copyFromLocalFile(new Path("D://op.log"), new Path(getHdfs().getWorkingDirectory().toString() + "/input/op.log"));
Job job = new Job(getConf(),"job 1");
job.setJarByClass(JobToSomeThing.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, getHdfsPath("input/op*.log"));
String outFileExt = "_" + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
FileOutputFormat.setOutputPath(job, getHdfsPath("out/helloworld"+outFileExt));
System.out.println(job.waitForCompletion(true));
getHdfs().copyToLocalFile(getHdfsPath("out/helloworld"+outFileExt), new Path("D://helloworld"+outFileExt));
}

是以单个job的方式,设置mapper类 reducer类,数据源来完成一个计算,把生成结果保存到hdfs的out/hellowrold下,最后我们把这个结果文件夹保存到我们本地D盘查看