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

下載Java2極速入門

徐老師·更新于2020-08- 上一節(jié)1剖析小文件問(wèn) 3YARN實(shí)戰(zhàn)下一我們知道MaRedce是分為Map階段和Reduce階段，其實(shí)提高執(zhí)行效率就是提高這兩個(gè)階段的執(zhí)行效率默認(rèn)情況下Map階段中Map任務(wù)的個(gè)數(shù)是和數(shù)據(jù)的InptSpit相關(guān)的，IntSplt的個(gè)數(shù)一般是和lck塊是有關(guān)聯(lián)的，所以可以認(rèn)為Map任務(wù)的個(gè)數(shù)和數(shù)據(jù)的block塊個(gè)數(shù)有關(guān)系，針對(duì)Map任務(wù)的個(gè)數(shù)我們一般是不需要干預(yù)的，除非是前面我們說(shuō)的海量小文件，那個(gè)時(shí)候可以考慮把小文件合并成大文件。其他情況是不需要調(diào)整的，那就剩下Reduce階段了，咱們前面說(shuō)過(guò)，默認(rèn)情況下reduce的個(gè)數(shù)是1個(gè)，所以現(xiàn)在MapRdce任務(wù)的壓力就集中在Reduce階段了，如果說(shuō)數(shù)據(jù)量比較大的時(shí)候，一個(gè)reduce任務(wù)處理起來(lái)肯定是比較慢的，所以我們可以考慮增加reduce任務(wù)的個(gè)數(shù)，這樣就可以實(shí)現(xiàn)數(shù)據(jù)分流了，提高計(jì)算效率了。job.getPartitionerClass方法看到默認(rèn)情況下會(huì)使用HashPartitioner這個(gè)分區(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，這里面其實(shí)就一個(gè)方法，getPartition，其實(shí)map里面每一條 (key.hashCode()&Integer.MAX_VALUE)%numReduceTasks的值，這個(gè)值默認(rèn)是1job.getNumReduceTasks()可知。所以最終任何值%1都返回0，那也就意味著他們都在0號(hào)分區(qū)，也就只有這一個(gè)分區(qū)。numReduceTasks的數(shù)目調(diào)大即可，這個(gè)其實(shí)就是reduce增加redcue任務(wù)個(gè)數(shù)在一定場(chǎng)景下是可以提高效率的，但是在一些特殊場(chǎng)景下單純?cè)黾觬educe任務(wù)個(gè)數(shù)是無(wú)法達(dá)到質(zhì)的提升的。其實(shí)在私底下我們是知道這份數(shù)據(jù)的大致情況的，這里面這1000w條數(shù)據(jù)，值為5的數(shù)據(jù)有910w條左右，剩下的9個(gè)數(shù)字一共只有90w條，那也就意味著，這份數(shù)據(jù)中，值為5的數(shù)據(jù)比較集中，或者說(shuō)值為5的數(shù)據(jù)屬于傾斜的數(shù)據(jù)，在這一整份數(shù)據(jù)中，它占得比重比其他的數(shù)據(jù)多得多。假設(shè)這100W條數(shù)據(jù)的文件有3個(gè)block，會(huì)產(chǎn)生3個(gè)Inptplt，最終會(huì)產(chǎn)生3個(gè)Map任務(wù)，默認(rèn)情況下只有一個(gè)reduce任務(wù)，所以所有的數(shù)據(jù)都會(huì)讓這一個(gè)reduce任務(wù)處理，這樣這個(gè)Reduce壓力肯定很大，大量的時(shí)間都消耗在了這里那根據(jù)我們前面的分析，我們可以增加reduce任務(wù)的數(shù)量，看下面這張圖，我們把reduce任務(wù)的數(shù)量調(diào)整到10個(gè)，這個(gè)時(shí)候就會(huì)把1000w條數(shù)據(jù)讓這10個(gè)reduce任務(wù)并行處理了，這個(gè)時(shí)候效率肯定會(huì)有一定的提升，但是最后我們會(huì)發(fā)現(xiàn)，性能提升是有限的，并沒(méi)有達(dá)到質(zhì)的提升，那這是為什么呢？我們來(lái)分析一下，剛才我們說(shuō)了我們這份數(shù)據(jù)中，值為5的數(shù)據(jù)有910w條，這就占了整份數(shù)據(jù)的90%了，那這90%的數(shù)據(jù)會(huì)被一個(gè)reduce任務(wù)處理，在這里假設(shè)是讓reduce5處理了，reduce5這個(gè)任務(wù)執(zhí)行的是比較慢的，其他reduce任務(wù)都執(zhí)行結(jié)束很長(zhǎng)時(shí)間了，它還沒(méi)執(zhí)行結(jié)束，因?yàn)閞educe5中處理的數(shù)據(jù)量和其他reduce中處理的數(shù)據(jù)量規(guī)模相差太大了，所以最終reduce5拖了后腿。咱們mapreduce任務(wù)執(zhí)行消耗的時(shí)間是一直統(tǒng)計(jì)到最后一個(gè)執(zhí)行結(jié)束的reduce任務(wù)，所以就算其他reduce任務(wù)早都執(zhí)行結(jié)束了也沒(méi)有用，整個(gè)mapreduce任務(wù)是沒(méi)有執(zhí)行結(jié)束的。MapRdce程序執(zhí)行時(shí)，Reduce節(jié)點(diǎn)大部分執(zhí)行完畢，但是有一個(gè)或者幾個(gè)Reduce節(jié)點(diǎn)運(yùn)行很慢，導(dǎo)致整個(gè)程序處理時(shí)間變得很長(zhǎng)增加reduce任務(wù)個(gè)數(shù)，這個(gè)屬于治標(biāo)不治本，針對(duì)傾斜不是太嚴(yán)重的數(shù)據(jù)是可以解決問(wèn)題的，針對(duì)傾斜嚴(yán)重的數(shù)據(jù)，這樣是解決不了根本問(wèn)題的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這個(gè)文件有點(diǎn)大，在wndws本地?zé)o法打開(kāi)，在這里我們?nèi)ヒ粭l數(shù)據(jù)看一下數(shù)據(jù)格式，前面是一個(gè)數(shù)字，后面是一行日志，這個(gè)數(shù)據(jù)是我自己造的，我們主要是使用前面的這個(gè)數(shù)字，后面的內(nèi)容主要是為了充數(shù)的，要不然文件太小，測(cè)試不出來(lái)效果。后面我們解析數(shù)據(jù)的時(shí)候只獲取前面這個(gè)數(shù)字即可，前面這個(gè)數(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下面我們來(lái)具體跑一個(gè)這份數(shù)據(jù)，首先復(fù)制一份WordCountJob的代碼，新的類名為1package2importimportimportimportimportimportimportimportimportimportimport15import17*數(shù)據(jù)傾斜-增加Reduce**CreatedbypublicclassWordCountJobSkew *MappublicstaticclassMyMapperextendsMapper<LongWritable,Loggerlogger=*需要實(shí)現(xiàn)map*這個(gè)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個(gè)reduce，所以一共有1行，在這我們截取了一部分，其中這里面有一個(gè)reduce任務(wù)消耗的時(shí)間比較長(zhǎng)，其他reduce任務(wù)的執(zhí)行時(shí)間都是~秒，這個(gè)reduce任務(wù)的執(zhí)行時(shí)間是分2秒，那就意味著值為的那9w數(shù)據(jù)進(jìn)入到這個(gè)ece了，所以它執(zhí)行的比較慢。針對(duì)這個(gè)操作我們需要去修改代碼，在這里我們?cè)僦匦聫?fù)制一個(gè)類，基于WordCountJobSkew1package2importimportimportimportimportimportimportimportimportimportimportimportimport *CreatedbypublicclassWordCountJobSkewRandKey *MappublicstaticclassMyMapperextendsMapper<LongWritable,Loggerlogger=Randomrandom=new*需要實(shí)現(xiàn)map*這個(gè)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=*針對(duì)<k2,{v2...}>的數(shù)據(jù)進(jìn)行累加求和，并且最終把數(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]));}