單元5 任務5.3-MapReduce完成電商銷售數據統計

《大數據平臺部署與運維》單元5MapReduce實現電商銷售數據統計任務5.3MapReduce完成電商銷售數據統計01掌握MapReduce自定義分區(qū)的概念與用法02掌握MapReduce自定義數據類型的概念與用法學習目標任務5.3MapReduce完成電商銷售數據統計【任務場景】經理：小張，接下來我們要使用MapReduce對電商銷售數據進行統計，可能要用到自定義分區(qū)和自定義數據類型，你了解MapReduce中的自定義分區(qū)和自定義數據類型嗎？小張：日常業(yè)務中Hadoop提供的數據類型有時不滿足使用，我們需要根據業(yè)務創(chuàng)建合適的自定義數據類型，也可以通過自定義分區(qū)對數據進行分區(qū)。經理：是的，自定義數據類型根據場景不同需要實現不同的接口，自定義分區(qū)可以把數據分到不同的reducer中。我給你一份后臺導出的數據，你用MapReduce統計一下每個買家收藏商品的數量，根據收藏日期做一下自定義分區(qū)。小張：好的。任務5.3MapReduce完成電商銷售數據統計【任務布置】MapReduce的運行依賴與JDK和Hadoop，因此必須將Hadoop的基礎環(huán)境提前安裝好，才能進行MapReduce的運行和操作。本任務要求在前面已經完成安裝部署Hadoop平臺的node1節(jié)點上完成。要求掌握MapReduce自定義分區(qū)與自定義數據類型的創(chuàng)建；基于IDEA進行開發(fā)，每個賣家收藏商品數量的統計，將收藏統計結果分為兩個分區(qū)，2020-04-14日之前及14日當天的數據為一個分區(qū)，2020-04-14日之后的數據為一個分區(qū)。任務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自定義分區(qū)1.

MapReduce

Partitioner類通過前面的學習我們知道Mapper最終處理的鍵值對<key,value>，是需要送到Reducer去合并的，合并的時候，有相同key的鍵/值對會送到同一個Reducer節(jié)點中進行歸并。哪個key到哪個Reducer的分配過程，是由Partitioner規(guī)定的。分區(qū)的目的是把具有相同key的值集合在一起，確保key相同的值都會在同一個reducer里面。這樣才能保證map的輸出數據被均勻的分發(fā)到reducer。HadoopMapReduce默認的HadoopPartitioner是哈希

Partitioner（HashPartitioner），它會對key計算哈希值，并基于該哈希值對鍵值對數據進行分區(qū)。任務5.3MapReduce完成電商銷售數據統計Partitioner的數量等于reducer的數量，Partitioner會根據reducer的數量來劃分數據，reducer數量可以通過下面的方法進行設置：JobConf.setNumReduceTasks()因此，來自同一個分區(qū)的數據會被一個reducer任務處理。需要注意的是，只有作業(yè)具有多個reducer任務時，分區(qū)才會被創(chuàng)建。也就是說，如果作業(yè)只有1個reducer任務，分區(qū)階段是不會發(fā)生的。低效的分區(qū)意味著，某些reducer將比其他reducer任務處理更多的數據。那么，整個作業(yè)的運行時間將取決于這些需要處理更多數據的reducer，也就是說，作業(yè)的運行時間會更長。為了克服低效分區(qū)的問題，我們可以自定義分區(qū)器（partitioner），這樣我們就可以根據具體業(yè)務修改分區(qū)邏輯，把數據均分的分發(fā)到不同的reducer任務里。任務5.3MapReduce完成電商銷售數據統計2.

Partitioner實現過程（1）先分析一下具體的業(yè)務邏輯，確定大概有多少個分區(qū)（2）首先書寫一個類，它要繼承

org.apache.hadoop.mapreduce.Partitioner這個抽象類（3）重寫publicintgetPartition這個方法，根據具體邏輯，讀數據庫或者配置返回相同的數字（4）在main方法中設置Partioner的類，job.setPartitionerClass(DataPartitioner.class);（5）設置Reducer的數量，job.setNumReduceTasks(2);任務5.3MapReduce完成電商銷售數據統計3.總結分區(qū)Partitioner主要作用在于以下兩點（1）根據業(yè)務需要，產生多個輸出文件；（2）多個reduce任務并發(fā)運行，提高整體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）數據類型，為了方便反射，必須要有一個無參的構造方法用來創(chuàng)建對象。（4）在自定義數據類型中，建議使用java的原生數據類型，最好不要使用Hadoop對原生類型進行封裝的數據類型。比如intx;//IntWritable和Strings;//Text等等。任務5.3MapReduce完成電商銷售數據統計【工作流程】創(chuàng)建新的項目；編寫自定義數據類型類，繼承Writable接口并重寫write和readFields方法；編寫Mapper類，繼承父類并重寫map方法；編寫自定義分區(qū)類，繼承父類并重寫getPartition方法；編寫Reducer類，繼承父類并重寫reduce方法；編寫驅動類，程序導出為jar包運行，查看結果。任務5.3MapReduce完成電商銷售數據統計【操作步驟】創(chuàng)建新的項目，項目類型選擇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.編寫自定義分區(qū)類，繼承父類并重寫getPartition方法importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.lib.partition.HashPartitioner;

publicclassMyPartitionextendsHashPartitioner<Text,CollectionWritable>{@OverridepublicintgetPartition(Textk2,CollectionWritablev2,intnumReduceTasks){Stringdate=v2.getCollectTime();//通過比對收藏時間進行分區(qū)

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);//設置自定義分區(qū)類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