单元5 任务5.3-MapReduce完成电商销售数据统计

《大数据平台部署与运维》单元5MapReduce实现电商销售数据统计任务5.3MapReduce完成电商销售数据统计01掌握MapReduce自定义分区的概念与用法02掌握MapReduce自定义数据类型的概念与用法学习目标任务5.3MapReduce完成电商销售数据统计【任务场景】经理：小张，接下来我们要使用MapReduce对电商销售数据进行统计，可能要用到自定义分区和自定义数据类型，你了解MapReduce中的自定义分区和自定义数据类型吗？小张：日常业务中Hadoop提供的数据类型有时不满足使用，我们需要根据业务创建合适的自定义数据类型，也可以通过自定义分区对数据进行分区。经理：是的，自定义数据类型根据场景不同需要实现不同的接口，自定义分区可以把数据分到不同的reducer中。我给你一份后台导出的数据，你用MapReduce统计一下每个买家收藏商品的数量，根据收藏日期做一下自定义分区。小张：好的。任务5.3MapReduce完成电商销售数据统计【任务布置】MapReduce的运行依赖与JDK和Hadoop，因此必须将Hadoop的基础环境提前安装好，才能进行MapReduce的运行和操作。本任务要求在前面已经完成安装部署Hadoop平台的node1节点上完成。要求掌握MapReduce自定义分区与自定义数据类型的创建；基于IDEA进行开发，每个卖家收藏商品数量的统计，将收藏统计结果分为两个分区，2020-04-14日之前及14日当天的数据为一个分区，2020-04-14日之后的数据为一个分区。任务5.3MapReduce完成电商销售数据统计5.3.1MapReduce完成电商销售数据统计的流程数据介绍现有某电商网站用户对商品的收藏数据，记录了用户收藏的商品id以及收藏日期，名为buyer_favorite1。buyer_favorite1包含：买家id，商品id，收藏日期这三个字段，数据以空格分割，样本数据及格式如下：1018110004812020-04-0416:54:312000110015972020-04-0715:07:522000110015602020-04-0715:08:272004210013682020-04-0808:20:302006710020612020-04-0816:45:332005610032892020-04-1210:50:552005610032902020-04-1211:57:352005610032922020-04-1212:05:292005410024202020-04-1415:24:12……任务5.3MapReduce完成电商销售数据统计5.3.2自定义分区1.

MapReduce

Partitioner类通过前面的学习我们知道Mapper最终处理的键值对<key,value>，是需要送到Reducer去合并的，合并的时候，有相同key的键/值对会送到同一个Reducer节点中进行归并。哪个key到哪个Reducer的分配过程，是由Partitioner规定的。分区的目的是把具有相同key的值集合在一起，确保key相同的值都会在同一个reducer里面。这样才能保证map的输出数据被均匀的分发到reducer。HadoopMapReduce默认的HadoopPartitioner是哈希

Partitioner（HashPartitioner），它会对key计算哈希值，并基于该哈希值对键值对数据进行分区。任务5.3MapReduce完成电商销售数据统计Partitioner的数量等于reducer的数量，Partitioner会根据reducer的数量来划分数据，reducer数量可以通过下面的方法进行设置：JobConf.setNumReduceTasks()因此，来自同一个分区的数据会被一个reducer任务处理。需要注意的是，只有作业具有多个reducer任务时，分区才会被创建。也就是说，如果作业只有1个reducer任务，分区阶段是不会发生的。低效的分区意味着，某些reducer将比其他reducer任务处理更多的数据。那么，整个作业的运行时间将取决于这些需要处理更多数据的reducer，也就是说，作业的运行时间会更长。为了克服低效分区的问题，我们可以自定义分区器（partitioner），这样我们就可以根据具体业务修改分区逻辑，把数据均分的分发到不同的reducer任务里。任务5.3MapReduce完成电商销售数据统计2.

Partitioner实现过程（1）先分析一下具体的业务逻辑，确定大概有多少个分区（2）首先书写一个类，它要继承

org.apache.hadoop.mapreduce.Partitioner这个抽象类（3）重写publicintgetPartition这个方法，根据具体逻辑，读数据库或者配置返回相同的数字（4）在main方法中设置Partioner的类，job.setPartitionerClass(DataPartitioner.class);（5）设置Reducer的数量，job.setNumReduceTasks(2);任务5.3MapReduce完成电商销售数据统计3.总结分区Partitioner主要作用在于以下两点（1）根据业务需要，产生多个输出文件；（2）多个reduce任务并发运行，提高整体job的运行效率任务5.3MapReduce完成电商销售数据统计5.3.3自定义数据类型Hadoop使用了自己写的序列化格式

Writable，它格式紧凑，速度快，但是它很难用Java以外的语言进行拓展或使用，因为Writable是Hadoop的核心，大多数MapReduce程序都会为键和值使用它，Hadoop中的数据类型都要实现Writable接口，以便用这些类型定义的数据可以被网络传输和文件存储。任务5.3MapReduce完成电商销售数据统计自定义数据类型实现过程（1）继承接口Writable,实现其方法write()和readFields(),以便该数据能被序列化后完成网络传输或文件输入/输出。（2）如果该数据需要作为主键key使用，或需要比较数值大小时，则需要实现WritalbeComparable接口,实现其方法write(),readFields(),CompareTo()。（3）数据类型，为了方便反射，必须要有一个无参的构造方法用来创建对象。（4）在自定义数据类型中，建议使用java的原生数据类型，最好不要使用Hadoop对原生类型进行封装的数据类型。比如intx;//IntWritable和Strings;//Text等等。任务5.3MapReduce完成电商销售数据统计【工作流程】创建新的项目；编写自定义数据类型类，继承Writable接口并重写write和readFields方法；编写Mapper类，继承父类并重写map方法；编写自定义分区类，继承父类并重写getPartition方法；编写Reducer类，继承父类并重写reduce方法；编写驱动类，程序导出为jar包运行，查看结果。任务5.3MapReduce完成电商销售数据统计【操作步骤】创建新的项目，项目类型选择Maven，项目名称为MRCollectionCount。1.首先编写自定义数据类型类：CollectionWritable，实现Writable接口并重写write和readFields方法。importorg.apache.hadoop.io.Writable;importjava.io.DataInput;importjava.io.DataOutput;importjava.io.IOException;publicclassCollectionWritableimplementsWritable{privateintcount;//收藏次数privateStringcollectTime;//收藏时间publicintgetCount(){returncount;}publicvoidsetCount(intcount){this.count=count;}publicStringgetCollectTime(){returncollectTime;}publicvoidsetCollectTime(StringcollectTime){this.collectTime=collectTime;}publicCollectionWritable(intcount,StringcollectTime){super();this.collectTime=collectTime;this.count=count;}

publicCollectionWritable(){

}@Overridepublicvoidwrite(DataOutputdataOutput)throwsIOException{dataOutput.writeInt(this.count);dataOutput.writeUTF(this.collectTime);}@OverridepublicvoidreadFields(DataInputdataInput)throwsIOException{this.count=dataInput.readInt();this.collectTime=dataInput.readUTF();}@OverridepublicStringtoString(){return"CollectionWritable{"+"count="+count+",collectTime='"+collectTime+'\''+'}';}}

任务5.3MapReduce完成电商销售数据统计2.编写Mapper类，继承父类并重写map方法importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Mapper;

importjava.io.IOException;

publicclassMyMapperextendsMapper<LongWritable,Text,Text,CollectionWritable>{@Overrideprotectedvoidmap(LongWritablekey,Textvalue,Mapper<LongWritable,Text,Text,CollectionWritable>.Contextcontext)throwsIOException,InterruptedException{//通过空格分割finalString[]splited=value.toString().split("");//第一列为用户IDfinalStringuserId=splited[0];finalTextk2=newText(userId);//收藏次数记为1finalintcount=1;//第三列为收藏时间finalStringcollectTime=splited[2];finalCollectionWritablev2=newCollectionWritable(count,collectTime);context.write(k2,v2);}}

任务5.3MapReduce完成电商销售数据统计3.编写自定义分区类，继承父类并重写getPartition方法importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.lib.partition.HashPartitioner;

publicclassMyPartitionextendsHashPartitioner<Text,CollectionWritable>{@OverridepublicintgetPartition(Textk2,CollectionWritablev2,intnumReduceTasks){Stringdate=v2.getCollectTime();//通过比对收藏时间进行分区

if(pareTo("2020-04-14")>0){return1;}else{return0;}}}任务5.3MapReduce完成电商销售数据统计4.编写Reducer类，继承父类并重写reduce方法importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Reducer;

importjava.io.IOException;

publicclassMyReducerextendsReducer<Text,CollectionWritable,Text,LongWritable>{@Overrideprotectedvoidreduce(Textk2,Iterable<CollectionWritable>values,Reducer<Text,CollectionWritable,Text,LongWritable>.Contextcontext)throwsIOException,InterruptedException{Textk3=k2;longsum=0;//收藏次数进行累加for(CollectionWritablecollect:values){sum+=collect.getCount();}

LongWritablev3=newLongWritable(sum);

context.write(k3,v3);}}

任务5.3MapReduce完成电商销售数据统计5.编写驱动类importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importjava.io.IOException;publicclassCollectionApp{publicstaticvoidmain(String[]args)throwsIOException,ClassNotFoundException,InterruptedException{Configurationconf=newConfiguration();Jobjob=Job.getInstance(conf,"collectionJob");job.setJarByClass(CollectionApp.class);PathfileIn=newPath(args[0]);PathfileOut=newPath(args[1]);job.setMapperClass(MyMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(CollectionWritable.class);FileInputFormat.setInputPaths(job,fileIn);//设置自定义分区类job.setPartitionerClass(MyPartition.class);//将reduce任务数量设置为2job.setNumReduceTasks(2);job.setReducerClass(MyReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(LongWritable.class);FileOutputFormat.setOutputPath(job,fileOut);Booleanr