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

这个主要就是使用了FlowBean.java中的代码来实现的，主要是继承了WritableComparable<FlowBean>接口来实现，然后重写了compareTo()方法。流量分析系统

@Override
	public int compareTo(FlowBean o) {
		
		return this.sumFlow>o.getSumFlow() ? -1:1;
	}
	

按照同样的方法对这个文件打成jar包，然后使用hadoop的相关语句进行执行就可以了。

bin/hadoop jar ../lx/flowsort.jar cn/tf/flow/FlowCountSort  /flow/output  /flow/sortoutput

结果图：

流量汇总之后的结果需要按照省份输出到不同的结果文件中，需要解决两个问题：

1、如何让mr的最终结果产生多个文件： 原理：MR中的结果文件数量由reduce

task的数量绝对，是一一对应的 做法：在代码中指定reduce task的数量

2、如何让手机号进入正确的文件 原理：让不同手机号数据发给正确的reduce task，就进入了正确的结果文件

要自定义MR中的分区partition的机制（默认的机制是按照kv中k的hashcode%reducetask数）

做法：自定义一个类来干预MR的分区策略——Partitioner的自定义实现类

主要代码与前面的排序是非常类的，只要在main方法中添加如下两行代码就可以了。

          //指定自定义的partitioner
		job.setPartitionerClass(ProvincePartioner.class);
		
		job.setNumReduceTasks(5);
		

这里我们需要新建一个ProvincePartioner.java来处理号码分类的逻辑。

public class ProvincePartioner extends Partitioner<Text, FlowBean>{
	
	
private static HashMap<String, Integer> provinceMap = new HashMap<String, Integer>();
	
	static {
		
		provinceMap.put("135", 0);
		provinceMap.put("136", 1);
		provinceMap.put("137", 2);
		provinceMap.put("138", 3);		
	}
	
	@Override
	public int getPartition(Text key, FlowBean value, int numPartitions) {

		String prefix = key.toString().substring(0, 3);
		Integer partNum = provinceMap.get(prefix);
		if(partNum == null) partNum=4;
		
		return partNum;
	}

}

执行方法和前面也是一样的。从执行的流程中我们可以看到这里启动了5个reduce task,因为我这里数据量比较小，所以只启动了一个map task。

到这里，整个用户流量分析系统就全部结束了。关于大数据的更多内容，欢迎关注。点击左上角头像下方“点击关注".感谢您的支持！

数据源下载地址：

源码项目地址：https://github.com/sdksdk0/HDFS_MapReduce

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