MapReduce实现手机上网日志分析（排序）

2016-12-13 510

一、背景

1.1 流程

　　实现排序，分组拍上一篇通过Partitioner实现了。

　　实现接口，自动产生接口方法，写属性，产生getter和setter，序列化和反序列化属性，写比较方法，重写toString，为了方便复制写够着方法，不过重写够着方法map里需要不停地new，发现LongWritable有set方法，text也有，可以用，产生默认够着方法。

	public void set(String account,double income,double expense,double surplus) {
		this.account = account;
		this.income = income;
		this.expense = expense;
		this.surplus = income-expense;
	}

1.2 数据集

为了和上一篇保在知识上持递进，数据及换了，名字没变。

　　下面是输出结果，其实mr也会自动排序，不过string按字典序排序了。

二、理论知识

　　字符串拼接，记得以前自己写过，现在拿出来看看，http://www.cnblogs.com/hxsyl/archive/2012/10/18/2729112.html

　　简单总结扩展如下：String是final的，不能改变也不能继承，因此在每次对 String 类型进行改变的时候其实都等同于生成了一个新的 String 对象，然后将指针指向新的 String 对象，所以经常改变内容的字符串最好不要用 String ，因为每次生成对象都会对系统性能产生影响，特别当内存中无引用对象多了以后， JVM 的 GC 就会开始工作，那速度是一定会相当慢的。

　　如果for循环1w次，这句 string += "hello";的过程相当于将原有的string变量指向的对象内容取出与"hello"作字符串相加操作再存进另一个新的String对象当中，再让string变量指向新生成的对象。反编译出的字节码文件可以很清楚地看出，每次循环会new出一个StringBuilder对象，然后进行append操作，最后通过toString方法返回String对象。也就是说这个循环执行完毕new出了10000个对象，试想一下，如果这些对象没有被回收，内存浪费不说，有可能重复使用赵成系统卡死。从上面还可以看出：string+="hello"的操作事实上会自动被JVM优化成：

　　StringBuilder str = new StringBuilder(string);

　　str.append("hello");

　　str.toString();

　　如果直接for循环里StringBuilder 的话会只是new一次。效率高。

　　而StringBuffer是线程安全的，多了synchronized关键字，也就是在多线程下会顺序读取换冲刺。

　参考了这个http://blog.csdn.net/loveyaozu/article/details/47037957

三、实体类

　　收入相同的话按消费从低到高，否则收入从高到低。

package cn.app.hadoop.mr.sort;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.math.BigDecimal;

import org.apache.hadoop.io.WritableComparable;
import org.apache.jasper.tagplugins.jstl.core.Out;

//Writable是序列化接口
//泛型是InfoBean，就像比较学生信息一样，成绩，性别等 ，封装在了一个bean里
//不过发现WritableComparable  有了序列化和反序列化
public class InfoBean implements WritableComparable<InfoBean>{
	
	
	private String account;
	//金钱类都需要BigDecimal，double顺势精度,不过不知道下边序列化咋写类型，所以先用double，估计writeUTF可以
	private double income;
	private double expense;
	private double surplus;
	
	
	public String getAccount() {
		return account;
	}
	public void setAccount(String account) {
		this.account = account;
	}
	public double getIncome() {
		return income;
	}
	public void setIncome(double income) {
		this.income = income;
	}
	public double getExpense() {
		return expense;
	}
	public void setExpense(double expense) {
		this.expense = expense;
	}
	public double getSurplus() {
		return surplus;
	}
	public void setSurplus(double surplus) {
		this.surplus = surplus;
	}
	public void readFields(DataInput in) throws IOException {
		// TODO Auto-generated method stub
		this.account = in.readUTF();
		this.income = in.readDouble();
		this.expense = in.readDouble();
		this.surplus = in.readDouble();
	}
	public void write(DataOutput out) throws IOException {
		// TODO Auto-generated method stub
		out.writeUTF(account);
		out.writeDouble(income);
		out.writeDouble(expense);
		out.writeDouble(surplus);
		
	}
	
	public void set(String account,double income,double expense) {
		this.account = account;
		this.income = income;
		this.expense = expense;
		this.surplus = income - expense;
	}
	

	public InfoBean() {
		super();
		// TODO Auto-generated constructor stub
	}
	@Override
	public String toString() {
		return "InfoBean [income=" + income + ", expense=" + expense
				+ ", surplus=" + surplus + "]";
	}
	public int compareTo(InfoBean o) {
		// TODO Auto-generated method stub
		if(this.income == o.getIncome()) {
			return this.expense>o.getExpense()?1:-1;
		}else {
			return this.income>o.getIncome()?-1:1;
		}
	}
}

四、第一种实现

4.1 Mapper

//第一个处理文本的话一般是LongWritable  或者object
//一行一行的文本是text
//输出的key的手机号 定位Text
//结果是DataBean  一定要实现Writable接口
public class InfoSortMapper extends Mapper<LongWritable, Text, Text, InfoBean> {

	
	private InfoBean v = new InfoBean();
	private Text k = new Text();
	
	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String line = value.toString();
		String[] fields = line.split("\t");
		String account = fields[0];
		double in = Double.parseDouble(fields[1]);
		double out = Double.parseDouble(fields[2]);
		
		//不用每次new  几遍不重写内存引用，也很站用资源
		k.set(account);
		v.set(account, in, out);
		
		context.write(k, v);
	}

　　4.2 Reducer

public class InfoSortReducer extends Reducer<Text, InfoBean, Text, InfoBean> {

	//k就是key，不需要
	private InfoBean v = new InfoBean();
	public void reduce(Text key, Iterable<InfoBean> value, Context context)
			throws IOException, InterruptedException {
		// process values
		double incomeSum = 0;
		double expenseSum = 0;
		for (InfoBean o : value) {
			incomeSum += o.getIncome();
			expenseSum += o.getExpense();
		}
		v.set(key.toString(), incomeSum, expenseSum);
		//databean会自动调用toString
		context.write(key,v);
	}
}

五、第二种实现

5.1 Mapper

//对 InfoBean  排序  k2就是他
public class SortMapper extends Mapper<LongWritable, Text, InfoBean, NullWritable> {

	
	private InfoBean k = new InfoBean();
	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String line = value.toString();
		String[] fields = line.split("\t");
		String account = fields[0];
		double in = Double.parseDouble(fields[1]);
		double out = Double.parseDouble(fields[2]);
		
		//不用每次new  几遍不重写内存引用，也很站用资源
		k.set(account, in, out);
		//value必须是NullWritable.get()，NullWritable不行，提示不是变量
		context.write(k, NullWritable.get());
	}
}

　　5.2 Reducer

//对 InfoBean  排序  k2就是他
public class SortMapper extends Mapper<LongWritable, Text, InfoBean, NullWritable> {

	
	private InfoBean k = new InfoBean();
	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String line = value.toString();
		String[] fields = line.split("\t");
		String account = fields[0];
		double in = Double.parseDouble(fields[1]);
		double out = Double.parseDouble(fields[2]);
		
		//不用每次new  几遍不重写内存引用，也很站用资源
		k.set(account, in, out);
		//value必须是NullWritable.get()，NullWritable不行，提示不是变量
		context.write(k, NullWritable.get());
	}
}

六、结束语

　　如果k2 v2和k4 v4，也就是mapp的输出和reducer的输出类型不一致的话必须在Main里也设置Mapper的输出,上面的第二种就是。

job.setMapOutputKeyClass(InfoBean.class);
		job.setMapOutputValueClass(NullWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(InfoBean.class);

　　否则java里不报错，加上log4j后看到类型不匹配。

微信关注我们

原文链接：https://yq.aliyun.com/articles/288401

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

2016Elastic中国开发者大会资料

2016Elastic中国开发者大会于2016年12月10日在京召开。笔者作为《Elasticsearch权威指南》的中文版自愿者翻译，有幸参加。本文将分享现场演讲资料和我做的笔记。大会介绍 Elastic 中国开发者大会 2016（Elastic{ON} Dev China 2016）是由 Elastic 官方在中国举办的第一次开发者大会，主要围绕 Elastic 的开源产品: Elasticsearch、Logstash、Kibana 和 Beats，探讨在搜索、数据实时分析、日志分析、安全等领域的实践与应用。大会邀请有来自Elastic、360、美团、58、苏宁等公司的技术专家带来的共计12个精彩的分享，与会人员大多为大数据相关的架构师、技术经理与一线工程师。大会亮点 1.Elastic 官方在中国召开的第一次开发者大会2.El

2016-12-12

843

Hadoop-2.7.0中HDFS NameNode HA实现综述

一、原理 HDFS中NameNode等的HA是基于ZooKeeper实现的。它应用了ZooKeeper集群的如下功能或特性： 1、只要半数以上节点还存活，就继续能对外提供服务； 2、ZooKeeper通过Paxos算法提供了leader选举功能，其它follower learn leader； 3、ZooKeeper提供了watcher机制，只要ZooKeeper上znode增减，或内容发生变化，或其子znode有增减，客户端都可以通过注册的watcher获得通知； 4、ZooKeeper提供了持久化节点和临时节点，尤其是临时节点EPHEMERAL，其在ZooKeeper客户端连接断掉后，会自动删除。正是基于ZooKeeper的上述特性，HDFS的NameNode实现了HA，NameNode的状态大致可分为Active和Standby两种，NameNode竞争在ZooKeeper指定路径上注册临时节点，将自己的host、port、nameserviceId、namenodeId等数据写入节点，哪个NameNode争先写入成功，哪个就成为Active Na...

2016-12-13

688

发表评论

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源，继承了IntelliJ IDEA强大的JS部分的功能。