2剖析數(shù)據(jù)傾斜問題與企業(yè)級解決方案教程

下載Java2極速入門

徐老師·更新于2020-08- 上一節(jié)1剖析小文件問 3YARN實戰(zhàn)下一我們知道MaRedce是分為Map階段和Reduce階段，其實提高執(zhí)行效率就是提高這兩個階段的執(zhí)行效率默認情況下Map階段中Map任務的個數(shù)是和數(shù)據(jù)的InptSpit相關的，IntSplt的個數(shù)一般是和lck塊是有關聯(lián)的，所以可以認為Map任務的個數(shù)和數(shù)據(jù)的block塊個數(shù)有關系，針對Map任務的個數(shù)我們一般是不需要干預的，除非是前面我們說的海量小文件，那個時候可以考慮把小文件合并成大文件。其他情況是不需要調(diào)整的，那就剩下Reduce階段了，咱們前面說過，默認情況下reduce的個數(shù)是1個，所以現(xiàn)在MapRdce任務的壓力就集中在Reduce階段了，如果說數(shù)據(jù)量比較大的時候，一個reduce任務處理起來肯定是比較慢的，所以我們可以考慮增加reduce任務的個數(shù)，這樣就可以實現(xiàn)數(shù)據(jù)分流了，提高計算效率了。job.getPartitionerClass方法看到默認情況下會使用HashPartitioner這個分區(qū)類/**Partitionkeysbytheir{@linkObject#hashCode()}.publicclassHashPartitioner<K,V>extendsPartitioner<K,V>5/**Use{@linkObject#hashCode()}topartition.publicintgetPartition(Kkey,VintnumReduceTasks)return(key.hashCode()&Integer.MAX_VALUE)% HashPartitioner繼承了Partitioner，這里面其實就一個方法，getPartition，其實map里面每一條 (key.hashCode()&Integer.MAX_VALUE)%numReduceTasks的值，這個值默認是1job.getNumReduceTasks()可知。所以最終任何值%1都返回0，那也就意味著他們都在0號分區(qū)，也就只有這一個分區(qū)。numReduceTasks的數(shù)目調(diào)大即可，這個其實就是reduce增加redcue任務個數(shù)在一定場景下是可以提高效率的，但是在一些特殊場景下單純增加reduce任務個數(shù)是無法達到質(zhì)的提升的。其實在私底下我們是知道這份數(shù)據(jù)的大致情況的，這里面這1000w條數(shù)據(jù)，值為5的數(shù)據(jù)有910w條左右，剩下的9個數(shù)字一共只有90w條，那也就意味著，這份數(shù)據(jù)中，值為5的數(shù)據(jù)比較集中，或者說值為5的數(shù)據(jù)屬于傾斜的數(shù)據(jù)，在這一整份數(shù)據(jù)中，它占得比重比其他的數(shù)據(jù)多得多。假設這100W條數(shù)據(jù)的文件有3個block，會產(chǎn)生3個Inptplt，最終會產(chǎn)生3個Map任務，默認情況下只有一個reduce任務，所以所有的數(shù)據(jù)都會讓這一個reduce任務處理，這樣這個Reduce壓力肯定很大，大量的時間都消耗在了這里那根據(jù)我們前面的分析，我們可以增加reduce任務的數(shù)量，看下面這張圖，我們把reduce任務的數(shù)量調(diào)整到10個，這個時候就會把1000w條數(shù)據(jù)讓這10個reduce任務并行處理了，這個時候效率肯定會有一定的提升，但是最后我們會發(fā)現(xiàn)，性能提升是有限的，并沒有達到質(zhì)的提升，那這是為什么呢？我們來分析一下，剛才我們說了我們這份數(shù)據(jù)中，值為5的數(shù)據(jù)有910w條，這就占了整份數(shù)據(jù)的90%了，那這90%的數(shù)據(jù)會被一個reduce任務處理，在這里假設是讓reduce5處理了，reduce5這個任務執(zhí)行的是比較慢的，其他reduce任務都執(zhí)行結束很長時間了，它還沒執(zhí)行結束，因為reduce5中處理的數(shù)據(jù)量和其他reduce中處理的數(shù)據(jù)量規(guī)模相差太大了，所以最終reduce5拖了后腿。咱們mapreduce任務執(zhí)行消耗的時間是一直統(tǒng)計到最后一個執(zhí)行結束的reduce任務，所以就算其他reduce任務早都執(zhí)行結束了也沒有用，整個mapreduce任務是沒有執(zhí)行結束的。MapRdce程序執(zhí)行時，Reduce節(jié)點大部分執(zhí)行完畢，但是有一個或者幾個Reduce節(jié)點運行很慢，導致整個程序處理時間變得很長增加reduce任務個數(shù)，這個屬于治標不治本，針對傾斜不是太嚴重的數(shù)據(jù)是可以解決問題的，針對傾斜嚴重的數(shù)據(jù)，這樣是解決不了根本問題的1[root@bigdata01soft]#2total3drwxr-xr-x.91001244Apr2620:34hadoop- -rw-r--r--.1rootroot345625475Jul192019hadoop- -rw-r--r--.1rootroot1860100000Apr2721:58 drwxr-xr-x. 10 245Dec162018-rw-r--r--.1rootroot194042837Apr623:14jdk-8u202-linux--rw-r--r--.1rootroot147616384Apr2716:22-rw-r--r--.1rootroot147976384Apr2716:22這個文件有點大，在wndws本地無法打開，在這里我們?nèi)ヒ粭l數(shù)據(jù)看一下數(shù)據(jù)格式，前面是一個數(shù)字，后面是一行日志，這個數(shù)據(jù)是我自己造的，我們主要是使用前面的這個數(shù)字，后面的內(nèi)容主要是為了充數(shù)的，要不然文件太小，測試不出來效果。后面我們解析數(shù)據(jù)的時候只獲取前面這個數(shù)字即可，前面這個數(shù)字是1-10之間的數(shù)字[root@bigdata01soft]#tail-110INFOmainorg.apache.hadoop.mapreduce.lib.output.FileOutputCommitter:[root@bigdata01soft]#hdfsdfs-puthello_10000000.dat[root@bigdata01soft]#hdfsdfs-ls 2rootsupergroup18601000002020-04-2722:01下面我們來具體跑一個這份數(shù)據(jù)，首先復制一份WordCountJob的代碼，新的類名為1package2importimportimportimportimportimportimportimportimportimportimport15import17*數(shù)據(jù)傾斜-增加Reduce**CreatedbypublicclassWordCountJobSkew *MappublicstaticclassMyMapperextendsMapper<LongWritable,Loggerlogger=*需要實現(xiàn)map*這個map函數(shù)就是可以接收<k1,v1>，產(chǎn)生*@param*@param*@param*@throws*@throwsprotectedvoidmap(LongWritablek1,Textv1,ContextthrowsIOException,InterruptedException//k1代表的是每一行數(shù)據(jù)的行首偏移量，v1String[]words=v1.toString().split("Textk2=newLongWritablev2=new *ReducepublicstaticclassMyReducerextendsLoggerlogger=

@param@param@param@throws@throwsprotectedvoidreduce(Textk2,Iterable<LongWritable>v2s,ContextcothrowsIOException,InterruptedException{longsum=0L;for(LongWritablev2v2s){sum+=v2.get();if(sum%200==0){}}Textk3k2;LongWritablev3=new}}*組裝publicstaticvoidmain(String[]args){}Configurationconf=newJobjob=}catch(Exception}}} [root@bigdata01hadoop-3.2.0]#hadoopjardb_hadoop-1.0-SNAPSHOT-jar-with- [root@bigdata01hadoop-3.2.0]#hdfsdfs-cat21324354657687989Elapsed:2mins,46sec[root@bigdata01[root@bigdata01hadoop-3.2.0]#hdfsdfs-rm-r[root@bigdata01hadoop-3.2.0]#hadoopjardb_hadoop-1.0-SNAPSHOT-jar-with-Elapsed:2mins,這里由于有0個reduce，所以一共有1行，在這我們截取了一部分，其中這里面有一個reduce任務消耗的時間比較長，其他reduce任務的執(zhí)行時間都是~秒，這個reduce任務的執(zhí)行時間是分2秒，那就意味著值為的那9w數(shù)據(jù)進入到這個ece了，所以它執(zhí)行的比較慢。針對這個操作我們需要去修改代碼，在這里我們再重新復制一個類，基于WordCountJobSkew1package2importimportimportimportimportimportimportimportimportimportimportimportimport *CreatedbypublicclassWordCountJobSkewRandKey *MappublicstaticclassMyMapperextendsMapper<LongWritable,Loggerlogger=Randomrandom=new*需要實現(xiàn)map*這個map函數(shù)就是可以接收<k1,v1>，產(chǎn)生*@param*@param*@param*@throws*@throwsprotectedvoidmap(LongWritablek1,Textv1,ContextthrowsIOException,InterruptedException//k1代表的是每一行數(shù)據(jù)的行首偏移量，v1String[]words=v1.toString().split("Stringkey=key= Textk2=newLongWritablev2=new *ReducepublicstaticclassMyReducerextendsLoggerlogger=*針對<k2,{v2...}>的數(shù)據(jù)進行累加求和，并且最終把數(shù)據(jù)轉(zhuǎn)化為k3,v3*@param*@param*@param*@throws*@throwsprotectedvoidreduce(Textk2,Iterable<LongWritable>v2s,ContextthrowsIOException,InterruptedExceptionlongsum=for(LongWritablev2:sum+=if(sum%200}}Textk3=LongWritablev3=new}}

*組裝publicstaticvoidmain(String[]{}Configurationconf=newJobjob=FileInputFormat.setInputPaths(job,newPath(args[0]));}