一脸懵逼学习Hadoop中的MapReduce程序中自定义分组的实现

时间：2021-09-13 04:11:22

1：首先搞好实体类对象：

　　write 是把每个对象序列化到输出流，readFields是把输入流字节反序列化，实现WritableComparable，Java值对象的比较：一般需要重写toString(),hashCode(),equals()方法

 package com.areapartition;

 import java.io.DataInput;

 import java.io.DataOutput;

 import java.io.IOException;

 import org.apache.hadoop.io.Writable;

 import org.apache.hadoop.io.WritableComparable;

 /***

  *

  * @author Administrator

  * 1:write 是把每个对象序列化到输出流

  * 2:readFields是把输入流字节反序列化

  * 3:实现WritableComparable

  *      Java值对象的比较：一般需要重写toString(),hashCode(),equals()方法

  *

  */

 public class FlowBean implements WritableComparable<FlowBean>{

     private String phoneNumber;//电话号码

     private long upFlow;//上行流量

     private long downFlow;//下行流量

     private long sumFlow;//总流量

     public String getPhoneNumber() {

         return phoneNumber;

     }

     public void setPhoneNumber(String phoneNumber) {

         this.phoneNumber = phoneNumber;

     }

     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(long sumFlow) {

         this.sumFlow = sumFlow;

     }

     //为了对象数据的初始化方便，加入一个带参的构造函数

     public FlowBean(String phoneNumber, long upFlow, long downFlow) {

         this.phoneNumber = phoneNumber;

         this.upFlow = upFlow;

         this.downFlow = downFlow;

         this.sumFlow = upFlow + downFlow;

     }

     //在反序列化时候，反射机制需要调用空参的构造函数，所以定义了一个空参的构造函数

     public FlowBean() {

     }

     //重写toString()方法

     @Override

     public String toString() {

         return "" + upFlow + "\t" + downFlow + "\t" + sumFlow + "";

     }

     //从数据流中反序列出对象的数据

     //从数据流中读取字段时必须和序列化的顺序保持一致

     @Override

     public void readFields(DataInput in) throws IOException {

         phoneNumber = in.readUTF();

         upFlow = in.readLong();

         downFlow = in.readLong();

         sumFlow = in.readLong(); 

     }

     //将对象数据序列化到流中

     @Override

     public void write(DataOutput out) throws IOException {

         out.writeUTF(phoneNumber);

         out.writeLong(upFlow);

         out.writeLong(downFlow);

         out.writeLong(sumFlow);

     }

     //流量比较的实现方法

     @Override

     public int compareTo(FlowBean o) {

         //大就返回-1，小于等于返回1，进行倒序排序

         return sumFlow > o.sumFlow ? - : ;

     }

 }

2：流量分区处理操作的步骤：

　　 2. 1：对流量原始日志进行流量统计，将不同的省份的用户统计结果输出到不同文件；

　　 2.2：需要自定义改造两个机制：

　　　　2.2.1：改造分区的逻辑，自定义一个partitioner

　　　　2.2.2：自定义reducer task的并发任务数

 package com.areapartition;

 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;

 /***

  * 流量分区处理操作

  * @author Administrator

  * 1:对流量原始日志进行流量统计，将不同的省份的用户统计结果输出到不同文件；

  * 2:需要自定义改造两个机制：

  *      2.1:改造分区的逻辑，自定义一个partitioner

  *      2.2:自定义reducer task的并发任务数

  */

 public class FlowSumArea {

     public static class FlowSumAreaMapper extends Mapper<LongWritable, Text, Text, FlowBean>{

         @Override

         protected void map(LongWritable key, Text value, Context context)

                 throws IOException, InterruptedException {

             //拿到一行数据

             String line = value.toString();

             //切分成各个字段

             String[] fields = StringUtils.split(line, "\t");

             //获取到我们需要的字段

             String phoneNumber = fields[];

             long up_flow = Long.parseLong(fields[]);

             long down_flow = Long.parseLong(fields[]);

             //封装成key-value并且输出

             context.write(new Text(phoneNumber), new FlowBean(phoneNumber, up_flow, down_flow));

         }

     }

     public static class FlowSumAreaReducer extends Reducer<Text, FlowBean, Text, FlowBean>{

         @Override

         protected void reduce(Text key, Iterable<FlowBean> values, Context context)

                 throws IOException, InterruptedException {

             //遍历求和

             long up_flowSum = ;

             long down_flowSum = ;

             for(FlowBean fb : values){

                 up_flowSum += fb.getUpFlow();

                 down_flowSum += fb.getDownFlow();

             }

             //封装成key-value并且输出

             context.write(key, new FlowBean(key.toString(),up_flowSum,down_flowSum));

         }

     }

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

         //创建配置文件

         Configuration conf = new Configuration();

         //获取一个作业

         Job job = Job.getInstance(conf);

         //设置整个job所用的那些类在哪个jar包

         job.setJarByClass(FlowSumArea.class);

         //本job使用的mapper和reducer的类

         job.setMapperClass(FlowSumAreaMapper.class);

         job.setReducerClass(FlowSumAreaReducer.class);

         //设置我们自定义的分组逻辑定义

         job.setPartitionerClass(AreaPartitioner.class);

         //指定mapper的输出数据key-value类型

         job.setMapOutputKeyClass(Text.class);

         job.setMapOutputValueClass(FlowBean.class);

         //指定reduce的输出数据key-value类型Text

         job.setOutputKeyClass(Text.class);

         job.setOutputValueClass(FlowBean.class);

         //设置reduce的任务并发数，应该跟分组的数量保持一致

         job.setNumReduceTasks();

         //指定要处理的输入数据存放路径

         //FileInputFormat是所有以文件作为数据源的InputFormat实现的基类，

         //FileInputFormat保存作为job输入的所有文件，并实现了对输入文件计算splits的方法。

         //至于获得记录的方法是有不同的子类——TextInputFormat进行实现的。

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

         //指定处理结果的输出数据存放路径

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

         //将job提交给集群运行

         //job.waitForCompletion(true);

         //正常执行成功返回0，否则返回1

         System.exit(job.waitForCompletion(true) ?  : );;

     }

 }

3：从key中拿到手机号，查询手机归属地字典，不同的省份返回不同的组号：

　　3.1：Partitioner是partitioner的基类，如果需要定制partitioner也需要继承该类。

　　3.2：HashPartitioner是mapreduce的默认partitioner。计算方法是 which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到当前的目的reducer。

 package com.areapartition;

 import java.util.HashMap;

 import org.apache.hadoop.mapreduce.Partitioner;

 public class AreaPartitioner<KEY,VALUE> extends Partitioner<KEY, VALUE>{

     private static HashMap<String, Integer> areaMap = new HashMap<String,Integer>();

     static{

         areaMap.put("", );

         areaMap.put("", );

         areaMap.put("", );

         areaMap.put("", );

         areaMap.put("", );

         areaMap.put("", );

     }

     @Override

     public int getPartition(KEY key, VALUE value, int numPartitions) {

         //从key中拿到手机号，查询手机归属地字典，不同的省份返回不同的组号

         Integer areaCoder = areaMap.get(key.toString().subSequence(, )) == null ?  : areaMap.get(key.toString().subSequence(, ));

         return areaCoder;

     }

 }

4：将打好的jar包上传到虚拟机上面：

然后启动搭建的集群start-dfs.sh，start-yarn.sh：

然后操作如下所示：

 [root@master hadoop]# hadoop jar flowarea.jar com.areapartition.FlowSumArea /flow/data /flow/areaoutput4

 // :: INFO client.RMProxy: Connecting to ResourceManager at master/192.168.0.55:

 // :: WARN mapreduce.JobSubmitter: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this.

 // :: INFO input.FileInputFormat: Total input paths to process :

 // :: INFO mapreduce.JobSubmitter: number of splits:

 // :: INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1506324201206_0004

 // :: INFO impl.YarnClientImpl: Submitted application application_1506324201206_0004

 // :: INFO mapreduce.Job: The url to track the job: http://master:8088/proxy/application_1506324201206_0004/

 // :: INFO mapreduce.Job: Running job: job_1506324201206_0004

 // :: INFO mapreduce.Job: Job job_1506324201206_0004 running in uber mode : false

 // :: INFO mapreduce.Job:  map % reduce %

 // :: INFO mapreduce.Job:  map % reduce %

 // :: INFO mapreduce.Job:  map % reduce %

 // :: INFO mapreduce.Job:  map % reduce %

 // :: INFO mapreduce.Job:  map % reduce %

 // :: INFO mapreduce.Job:  map % reduce %

 // :: INFO mapreduce.Job:  map % reduce %

 // :: INFO mapreduce.Job:  map % reduce %

 // :: INFO mapreduce.Job: Job job_1506324201206_0004 completed successfully

 // :: INFO mapreduce.Job: Counters:

     File System Counters

         FILE: Number of bytes read=

         FILE: Number of bytes written=

         FILE: Number of read operations=

         FILE: Number of large read operations=

         FILE: Number of write operations=

         HDFS: Number of bytes read=

         HDFS: Number of bytes written=

         HDFS: Number of read operations=

         HDFS: Number of large read operations=

         HDFS: Number of write operations=

     Job Counters

         Launched map tasks=

         Launched reduce tasks=

         Data-local map tasks=

         Total time spent by all maps in occupied slots (ms)=

         Total time spent by all reduces in occupied slots (ms)=

         Total time spent by all map tasks (ms)=

         Total time spent by all reduce tasks (ms)=

         Total vcore-seconds taken by all map tasks=

         Total vcore-seconds taken by all reduce tasks=

         Total megabyte-seconds taken by all map tasks=

         Total megabyte-seconds taken by all reduce tasks=

     Map-Reduce Framework

         Map input records=

         Map output records=

         Map output bytes=

         Map output materialized bytes=

         Input split bytes=

         Combine input records=

         Combine output records=

         Reduce input groups=

         Reduce shuffle bytes=

         Reduce input records=

         Reduce output records=

         Spilled Records=

         Shuffled Maps =

         Failed Shuffles=

         Merged Map outputs=

         GC time elapsed (ms)=

         CPU time spent (ms)=

         Physical memory (bytes) snapshot=

         Virtual memory (bytes) snapshot=

         Total committed heap usage (bytes)=

     Shuffle Errors

         BAD_ID=

         CONNECTION=

         IO_ERROR=

         WRONG_LENGTH=

         WRONG_MAP=

         WRONG_REDUCE=

     File Input Format Counters

         Bytes Read=

     File Output Format Counters

         Bytes Written=

 [root@master hadoop]# hadoop fs -ls /flow/

 Found  items

 drwxr-xr-x   - root supergroup           -- : /flow/areaoutput

 drwxr-xr-x   - root supergroup           -- : /flow/areaoutput2

 drwxr-xr-x   - root supergroup           -- : /flow/areaoutput3

 drwxr-xr-x   - root supergroup           -- : /flow/areaoutput4

 -rw-r--r--    root supergroup        -- : /flow/data

 drwxr-xr-x   - root supergroup           -- : /flow/output

 drwxr-xr-x   - root supergroup           -- : /flow/output2

 drwxr-xr-x   - root supergroup           -- : /flow/output3

 drwxr-xr-x   - root supergroup           -- : /flow/output4

 drwxr-xr-x   - root supergroup           -- : /flow/sortoutput

 [root@master hadoop]# hadoop fs -ls /flow/areaoutput4

 Found  items

 -rw-r--r--    root supergroup           -- : /flow/areaoutput4/_SUCCESS

 -rw-r--r--    root supergroup          -- : /flow/areaoutput4/part-r-

 -rw-r--r--    root supergroup          -- : /flow/areaoutput4/part-r-

 -rw-r--r--    root supergroup         -- : /flow/areaoutput4/part-r-

 -rw-r--r--    root supergroup          -- : /flow/areaoutput4/part-r-

 -rw-r--r--    root supergroup         -- : /flow/areaoutput4/part-r-

 -rw-r--r--    root supergroup          -- : /flow/areaoutput4/part-r-

 -rw-r--r--    root supergroup         -- : /flow/areaoutput4/part-r-

 [root@master hadoop]# hadoop fs -cat /flow/areaoutput4/part-r-

 [root@master hadoop]# hadoop fs -cat /flow/areaoutput4/part-r-

 [root@master hadoop]# hadoop fs -cat /flow/areaoutput4/part-r-

 [root@master hadoop]# hadoop fs -cat /flow/areaoutput4/part-r-

 [root@master hadoop]# hadoop fs -cat /flow/areaoutput4/part-r-

 [root@master hadoop]# hadoop fs -cat /flow/areaoutput4/part-r-

 [root@master hadoop]# hadoop fs -cat /flow/areaoutput4/part-r-

5：复制多份测试数据操作如下，测试map的多线程执行：

　　5.1：map task 的并发数是切片的数量决定的，有多少个切片，就启动多少个map task。

　　5.2：切片是一个逻辑的概念，指的就是文件中数据的偏移量的范围。

　　5.3：切片的具体大小应该根据所处理的文件的大小来调整。

[root@master hadoop]# hadoop fs -mkdir /flow/data/

[root@master hadoop]# hadoop fs -put HTTP_20130313143750.dat /flow/data/

[root@master hadoop]# hadoop fs -cp /flow/data/HTTP_20130313143750.dat /flow/data/HTTP_20130313143750.dat.

[root@master hadoop]# hadoop fs -cp /flow/data/HTTP_20130313143750.dat /flow/data/HTTP_20130313143750.dat.

[root@master hadoop]# hadoop fs -cp /flow/data/HTTP_20130313143750.dat /flow/data/HTTP_20130313143750.dat.

[root@master hadoop]# hadoop fs -ls /flow/data/

Found  items

-rw-r--r--    root supergroup        -- : /flow/data/HTTP_20130313143750.dat

-rw-r--r--    root supergroup        -- : /flow/data/HTTP_20130313143750.dat.

-rw-r--r--    root supergroup        -- : /flow/data/HTTP_20130313143750.dat.

-rw-r--r--    root supergroup        -- : /flow/data/HTTP_20130313143750.dat.

[root@master hadoop]#

6：Combiners编程

　　6.1：每一个map可能会产生大量的输出，combiner的作用就是在map端对输出先做一次合并，以减少传输到reducer的数据量。

　　6.2：combiner最基本是实现本地key的归并，combiner具有类似本地的reduce功能。

　　6.3： 如果不用combiner，那么，所有的结果都是reduce完成，效率会相对低下。使用combiner，先完成的map会在本地聚合，提升速度。

　　6.4：注意：Combiner的输出是Reducer的输入，如果Combiner是可插拔的，添加Combiner绝不能改变最终的计算结果。所以Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。

7：shuffle机制：

　　 7.1：每个map有一个环形内存缓冲区，用于存储任务的输出。默认大小100MB（io.sort.mb属性），一旦达到阀值0.8（io.sort.spill.percent）,一个后台线程把内容写到(spill)磁盘的指定目录（mapred.local.dir）下的新建的一个溢出写文件。

　　 7.2：写磁盘前，要partition（分组）,sort（排序）。如果有combiner，combine排序后数据。

　　 7.3：等最后记录写完，合并全部溢出写文件为一个分区且排序的文件。

　　 7.4：Reducer通过Http方式得到输出文件的分区。

　　 7.5：TaskTracker为分区文件运行Reduce任务。复制阶段把Map输出复制到Reducer的内存或磁盘。一个Map任务完成，Reduce就开始复制输出。

　　 7.6：排序阶段合并map输出。然后走Reduce阶段。

相关文章