大数据培训之WritableComparable排序

WritableComparable排序

大数据培训

大数据培训技术

1.排序的分类

大数据培训

2.自定义排序WritableComparable

(1)原理分析

bean对象做为key传输,需要实现WritableComparable接口重写compareTo方法,就可以实现排序。

@Override

public int compareTo(FlowBean o) {

 

   int result;

 

   // 按照总流量大小,倒序排列

   if (sumFlow > bean.getSumFlow()) {

      result = -1;

   }else if (sumFlow < bean.getSumFlow()) {

      result = 1;

   }else {

      result = 0;

   }

 

   return result;

}

 WritableComparable排序案例实操(全排序)

1.需求

根据案例2.3产生的结果再次对总流量进行排序。

(1)输入数据

原始数据                          第一次处理后的数据

 

(2)期望输出数据

13509468723 7335       110349   117684

13736230513 2481       24681     27162

13956435636 132         1512       1644

13846544121 264         0            264

。。。 。。。

2.需求分析

大数据培训

3.代码实现

(1)FlowBean对象在在需求1基础上增加了比较功能

package com.atguigu.mapreduce.sort;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

 

public class FlowBean implements WritableComparable<FlowBean> {

 

  private long upFlow;

  private long downFlow;

  private long sumFlow;

 

  // 反序列化时,需要反射调用空参构造函数,所以必须有

  public FlowBean() {

     super();

  }

 

  public FlowBean(long upFlow, long downFlow) {

     super();

     this.upFlow = upFlow;

     this.downFlow = downFlow;

     this.sumFlow = upFlow + downFlow;

  }

 

  public void set(long upFlow, long downFlow) {

     this.upFlow = upFlow;

     this.downFlow = downFlow;

     this.sumFlow = upFlow + downFlow;

  }

 

  public long getSumFlow() {

     return sumFlow;

  }

 

  public void setSumFlow(long sumFlow) {

     this.sumFlow = sumFlow;

  }                     

 

  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;

  }

 

  /**

   * 序列化方法

   * @param out

   * @throws IOException

   */

  @Override

  public void write(DataOutput out) throws IOException {

     out.writeLong(upFlow);

     out.writeLong(downFlow);

     out.writeLong(sumFlow);

  }

 

  /**

   * 反序列化方法 注意反序列化的顺序和序列化的顺序完全一致

   * @param in

   * @throws IOException

   */

  @Override

  public void readFields(DataInput in) throws IOException {

     upFlow = in.readLong();

     downFlow = in.readLong();

     sumFlow = in.readLong();

  }

 

  @Override

  public String toString() {

     return upFlow + “\t” + downFlow + “\t” + sumFlow;

  }

 

  @Override

  public int compareTo(FlowBean o) {

 

     int result;

 

     // 按照总流量大小,倒序排列

     if (sumFlow > bean.getSumFlow()) {

        result = -1;

     }else if (sumFlow < bean.getSumFlow()) {

        result = 1;

     }else {

        result = 0;

     }

 

     return result;

  }

}

       (2)编写Mapper类

package com.atguigu.mapreduce.sort;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

 

public class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text>{

 

  FlowBean bean = new FlowBean();

  Text v = new Text();

 

  @Override

  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

 

     // 1 获取一行

     String line = value.toString();

 

     // 2 截取

     String[] fields = line.split(“\t”);

 

     // 3 封装对象

     String phoneNbr = fields[0];

     long upFlow = Long.parseLong(fields[1]);

     long downFlow = Long.parseLong(fields[2]);

 

     bean.set(upFlow, downFlow);

     v.set(phoneNbr);

 

     // 4 输出

     context.write(bean, v);

  }

}

       (3)编写Reducer类

package com.atguigu.mapreduce.sort;

import java.io.IOException;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

public class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean>{

 

  @Override

  protected void reduce(FlowBean key, Iterable<Text> values, Context context)  throws IOException, InterruptedException {

 

     // 循环输出,避免总流量相同情况

     for (Text text : values) {

        context.write(text, key);

     }

  }

}

       (4)编写Driver类

package com.atguigu.mapreduce.sort;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 

public class FlowCountSortDriver {

 

  public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {

 

     // 输入输出路径需要根据自己电脑上实际的输入输出路径设置

     args = new String[]{“e:/output1″,”e:/output2”};

 

     // 1 获取配置信息,或者job对象实例

     Configuration configuration = new Configuration();

     Job job = Job.getInstance(configuration);

 

     // 2 指定本程序的jar包所在的本地路径

     job.setJarByClass(FlowCountSortDriver.class);

 

     // 3 指定本业务job要使用的mapper/Reducer业务类

     job.setMapperClass(FlowCountSortMapper.class);

     job.setReducerClass(FlowCountSortReducer.class);

 

     // 4 指定mapper输出数据的kv类型

     job.setMapOutputKeyClass(FlowBean.class);

     job.setMapOutputValueClass(Text.class);

 

     // 5 指定最终输出的数据的kv类型

     job.setOutputKeyClass(Text.class);

     job.setOutputValueClass(FlowBean.class);

 

     // 6 指定job的输入原始文件所在目录

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

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

 

     // 7 将job中配置的相关参数,以及job所用的java类所在的jar包, 提交给yarn去运行

     boolean result = job.waitForCompletion(true);

     System.exit(result ? 0 : 1);

  }

}


上一篇:
下一篇:
关于尚硅谷
教育理念
名师团队
学员心声
资源下载
视频下载
资料下载
工具下载
加入我们
招聘岗位
岗位介绍
招贤纳师
联系我们
电话:010-56253825
邮箱:info@atguigu.com
地址:北京市昌平区宏福科技园综合楼6层(北京校区)

 深圳市宝安区西部硅谷大厦B座C区一层(深圳校区)

上海市松江区谷阳北路166号大江商厦6层(上海校区)