个人学习整理,所有资料来自尚硅谷
B站学习连接:添加链接描述
百度网盘连接:添加链接描述
序列化就是把内存中的对象,转换成字符序列(或其他数据传输协议)以便于存储到磁盘(持久化)和网络传输;
反序列化就是将收到字节序列(或其他数据传输协议)或者是磁盘的持久化数据,转换成内存中的对象。
1.1.1为什么要序列化 序列化可以存储“活的”对象,将“活的”对象发送到远程计算机。
1.1.3 为什么不用JAVA的序列化、 Java的序列化是一个重量级序列化框架,会附带很多额外的信息,不便于网络中高效 传输,所以Hadoop开发了一套序列化机制(Writable)。
1.1.4 Hadoop序列化特点:- 紧凑:高效实用存储空间
- 快速:读写数据的额外开销小
- 互 *** 作:支持多种语言的交互
具体实现bean对象序列化步骤需要如下7步:
(1)必须实现Writable接口
(2)反序列化时,需要反射调用空参构造函数,所以必须有空参构造
public FlowBean(){ super(); }
@Override public void write(DataOutput dataoutput) throws IOException{ dataoutput.writeLong(upFlow); dataoutput.writeLong(downFlow); dataoutput.writeLong(sumFlow); }
(4)重写反序列方法(顺序要和序列化方法一致)
@Override public void readFields(DataInput dataInput) throws IOException{ this.upFlow = dataInput.readLong(); this.downFlow = dataInput.readLong(); this.sumFlow = dataInput.readLong(); }
(5)反序列化的顺序和序列化的顺序完全一致
(6)要想把结果显示在文件中,需要重写toString(),可用"t"分开,方便后续用
(7)如果需要将自定义的bean放在key中传输,则还需要实现Comparable接口,因为MapReduce框架中的Shuffle过程要求对key必须能排序。
@Override public int comparaeTo(FlowBean o){ //倒序排序,从大到小 return this.sumFlow > o.getSumFlow() ? -1:1; }1.3 序列化案例实 ***
需求:统计每一个手机号耗费的总上行流量、总下行流量、总流量
输入数据格式:
期望输出数据格式:
需求分析:
public class FlowBean implements Writable {//实现Writable接口 private long upFlow; private long downFlow; private long sumFlow; //反序列化时,需要反射调用空参构造函数,所以必须有空参构造 public FlowBean(){ } public long getUpFlow(){return upFlow;} public void setUpFlow(long upFlow){this.upFlow=upFlow;} public long getDownFlow(){return downFlow;} public void setDownFlow(long downFlow){this.downFlow=downFlow;} public long getSumFlow(){return sumFlow;} public void setSumFlow(){this.sumFlow=this.upFlow+this.downFlow;} //重写序列化方法 @Override public void write(DataOutput dataOutput) throws IOException{ dataOutput.writeLong(upFlow); dataOutput.writeLong(downFlow); dataOutput.wirteLong(sumFlow); } //重写反序列化方法 @Override public void readFields(DataInput dataInput) throws IOException{ this.upFlow = dataInput.readLong(); this.downFlow = dataInput.readLong(); this.sumFlow = dataInput.readLong(); } //重写toString方法 @Override public String toString(){ return upFlow + "t" + downFlow + "t" + sumFlow; } }2.编写Mapper类
public class FlowMapper extends Mapper3.重写Reducer类{ private Text outK = new Text(); private FlowBean outV = new FlowBean(); @Override protected void map(LongWritable key,Text value,Mappe .Context context) throws IOException,InterruptedException{ //获取一行 String line = value.toString(); //切割 String[] split = line.split("t"); //抓取想要的数据(手机号,上行/下行流量) String phone = split[1]; String up = split[split.length-3]; String down = split[split.length-2]; //封装 outK.set(phone); outV.setUpFlow(Long.parseLong(up)); outV.setDownFlow(Long.parseLong(down)); outV.setSumFlow(); //写出 context.write(outK,outV); } }
public class FlowReducer extends Reducer4.编写Driver类{ private FlowBean outV = new FlowBean(); @Override protected void reduce(Text key,Iterable values,Reducer .Context context) throws IOException, InterruptedException { //遍历集合累加 long totalUp = 0; long totalDown = 0; for (FlowBean value:values){ totalUp += value.getUpFlow(); totalDown += value.getDownFlow(); } //封装 outV.setUpFlow(totalUp); outV.setDownFlow(totalDown); outV.setSumFlow(); //写出 context.write(key,outV); } }
public class FlowDriver{ public static void main(String[] args) throws OException, InterruptedException, ClassNotFoundException{ //1、获取配置信息以及获取job对象 Configuration conf = new Configuration(); Job job = Job.getInstance(conf); //2、关联本Driver的jar job.setJarByClass(FlowDriver.class); //3、关联Mapper和Reducer的jar job.setMapperClass(FlowMapper.class); job.setReducerClass(FlowReducer.class); //4、设置Mapper输出的kv类型 job.setMapOutKeyClass(Text.class); job.setMapOutValueClass(FlowBean.class); //5、设置最终输出的kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(FlowBean.class); //6、设置输入和输出路径 FileInputFormat.setInputPaths(job,new Path(args[0])); FileOutputFormat.setOutputPath(job,new Path(args[1])); //7、提交job boolean result = job.waitForComplete(true); System.exit(result?0:1);//为0时正常退出程序 } }
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)