流量分析系统_统计分析系统的作用_sdn流量分析(2)

然后就是这个统计的代码了，新建一个FlowCount.java.在这个类里面，我直接把Mapper和Reduce写在同一个类里面了，如果按规范的要求应该是要分开写的。

在mapper中，获取后面三段数据的，所以我的这里length-2,length-3.

 bin/hadoop fs -mkdir -p /flow/data
 bin/hadoop fs -put HTTP_20130313143750.dat /flow/data
 bin/hadoop jar  ../lx/flow.jar

把上面这个MapReduce工程打包成一个jar文件，然后用hadoop来执行这个jar文件。例如我放在~/hadoop/lx/flow.jar，然后再hadoop安装目录中执行

最后执行结果如下：

如果你上面这个基本操作以及完成了的话，按总流量排序就非常简单了。我们新建一个FlowCountSort.java.

全部代码如下：

package cn.tf.flow;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;

public class FlowCountSort {

public static class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text>{
		
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			
			String line=value.toString();
			String[] fields=StringUtils.split(line,"\t");
			
			String phone=fields[0];
			long upSum=Long.parseLong(fields[1]);
			long dSum=Long.parseLong(fields[2]);
			
			FlowBean sumBean=new FlowBean(upSum,dSum);
			
			context.write(sumBean, new Text(phone));
		
		}	
}

	public static class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean>{
		
		//进来的“一组”数据就是一个手机的流量bean和手机号
		@Override
		protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
	
			context.write(values.iterator().next(), key);
		}
	}

	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(FlowCountSort.class);

		job.setMapperClass(FlowCountSortMapper.class);
		job.setReducerClass(FlowCountSortReducer.class);

		job.setMapOutputKeyClass(FlowBean.class);
		job.setMapOutputValueClass(Text.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		boolean res = job.waitForCompletion(true);
		System.exit(res ? 0 : 1);

	}
	
}

本文来自电脑杂谈，转载请注明本文网址：
http://www.pc-fly.com/a/tongxinshuyu/article-39516-2.html