典型的業(yè)務(wù)場景

典型的業(yè)務(wù)場景2學(xué)習(xí)任務(wù)了解讀文件時避免數(shù)據(jù)源的數(shù)據(jù)傾斜了解調(diào)整并行度分散同一個Task3知識目標(biāo)數(shù)據(jù)傾斜常見種類有哪幾種常見數(shù)據(jù)傾斜的常見特征有哪些數(shù)據(jù)傾斜解決方案大致分為哪幾步01能力目標(biāo)了解讀文件時避免數(shù)據(jù)源的數(shù)據(jù)傾斜能夠簡單敘述調(diào)整并行度分散同一個Task的不同Key02學(xué)習(xí)目標(biāo)4目錄01避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件02調(diào)整并行度分散同一個Task的不同Key5避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件Spark以通過textFile(path,minPartitions)方法讀取文件時，使用TextFileFormat。對于不可切分的文件，每個文件對應(yīng)一個Split從而對應(yīng)一個Partition。此時各文件大小是否一致，很大程度上決定了是否存在數(shù)據(jù)源的數(shù)據(jù)傾斜。另外，對于不可切分的壓縮文件，即使壓縮后的文件大小一致，它所包含的實(shí)際數(shù)據(jù)量也可能差別很多，因為源文件數(shù)據(jù)重復(fù)度越高，壓縮比越高。此時可通過在數(shù)據(jù)生成端將不可切分文件存儲為可切分文件，或者保證各文件包含數(shù)據(jù)量相同的方式避免數(shù)據(jù)傾斜。原理6避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件對于可切分的文件，每個Split大小由如下算法決定。其中g(shù)oalSize等于所有文件總大小除以minPartitions。而blockSize，如果是HDFS文件，由文件本身的block大小決定；如果是Linux本地文件，且使用本地模式，由fs.local.block.size決定。原理protectedlongcomputeSplitSize(longgoalSize,longminSize,longblockSize){returnMath.max(minSize,Math.min(goalSize,blockSize));}默認(rèn)情況下各Split的大小不會太大，一般相當(dāng)于一個Block大小（在Hadoop2中，默認(rèn)值為128MB），所以數(shù)據(jù)傾斜問題不明顯。如果出現(xiàn)了嚴(yán)重的數(shù)據(jù)傾斜，可通過上述參數(shù)調(diào)整。7避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件現(xiàn)通過腳本生成一些文本文件，并通過如下代碼進(jìn)行簡單的單詞計數(shù)。為避免Shuffle，只計單詞總個數(shù)，不須對單詞進(jìn)行分組計數(shù)。案例SparkConfsparkConf=newSparkConf().setAppName("ReadFileSkewDemo");JavaSparkContextjavaSparkContext=newJavaSparkContext(sparkConf);longcount=javaSparkContext.textFile(inputFile,minPartitions) .flatMap((Stringline)->Arrays.asList(line.split("")).iterator()).count();System.out.printf("totalwords:%s",count);javaSparkContext.stop();8避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件總共生成如下11個csv文件，其中10個大小均為271.9MB，另外一個大小為8.5GB。案例9避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件之后將8.5GB大小的文件使用gzip壓縮，壓縮后大小僅為25.3MB。案例10避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件使用如上代碼對未壓縮文件夾進(jìn)行單詞計數(shù)操作。Split大小為max(minSize,min(goalSize,blockSize)=max(1B,min((271.910+8.51024)/1MB,128MB)=128MB,無明顯數(shù)據(jù)傾斜。案例11避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件使用同樣代碼對包含壓縮文件的文件夾進(jìn)行同樣的單詞計數(shù)操作。未壓縮文件的Split大小仍然為128MB，而壓縮文件（gzip壓縮）由于不可切分，且大小僅為25.3MB，因此該文件作為一個單獨(dú)的Split/Partition。雖然該文件相對較小，但是它由8.5GB文件壓縮而來，包含數(shù)據(jù)量是其它未壓縮文件的32倍，因此處理該Split/Partition/文件的Task耗時為4.4分鐘，遠(yuǎn)高于其它Task的10秒。運(yùn)行結(jié)果如下圖所示。案例12避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件案例13避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件由于上述gzip壓縮文件大小為25.3MB，小于128MB的Split大小，不能證明gzip壓縮文件不可切分?，F(xiàn)將minPartitions從默認(rèn)的1設(shè)置為229，從而目標(biāo)Split大小為max(minSize,min(goalSize,blockSize)=12MB。如果gzip壓縮文件可切分，則所有Split/Partition大小都不會遠(yuǎn)大于12。反之，如果仍然存在25.3MB的Partition，則說明gzip壓縮文件確實(shí)不可切分，在生成不可切分文件時需要如上文所述保證各文件數(shù)量大致相同。案例14避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件如下圖所示，gzip壓縮文件對應(yīng)的Split/Partition大小為25.3MB，其它Split大小均為12MB左右。而該Task耗時4.7分鐘，遠(yuǎn)大于其它Task的4秒。案例15避免數(shù)據(jù)源的數(shù)據(jù)傾斜--讀文件適用場景：數(shù)據(jù)源存在不可切分文件，且文件內(nèi)包含的數(shù)據(jù)量相差較大。解決方案：盡量使用可切分的格式代替不可切分的格式，或者保證各文件實(shí)際包含數(shù)據(jù)量大致相同。優(yōu)勢：可撤底消除數(shù)據(jù)源的數(shù)據(jù)傾斜，效果顯著。劣勢：數(shù)據(jù)源一般來源于外部系統(tǒng)，需要外部系統(tǒng)的支持。小結(jié)16調(diào)整并行度分散同一個Task的不同KeySpark在做Shuffle時，默認(rèn)使用HashPartitioner（非HashShuffle）對數(shù)據(jù)進(jìn)行分區(qū)。如果并行度設(shè)置的不合適，可能造成大量不相同的Key對應(yīng)的數(shù)據(jù)被分配到了同一個Task上，造成該Task所處理的數(shù)據(jù)遠(yuǎn)大于其它Task，從而造成數(shù)據(jù)傾斜。如果調(diào)整Shuffle時的并行度，使得原本被分配到同一Task的不同Key發(fā)配到不同Task上處理，則可降低原Task所需處理的數(shù)據(jù)量，從而緩解數(shù)據(jù)傾斜問題造成的短板效應(yīng)。原理17調(diào)整并行度分散同一個Task的不同Key現(xiàn)有一張測試表，名為student_external，內(nèi)有10.5億條數(shù)據(jù)，每條數(shù)據(jù)有一個唯一的id值?，F(xiàn)從中取出id取值為9億到10.5億的共1.5億條數(shù)據(jù)，并通過一些處理，使得id為9億到9.4億間的所有數(shù)據(jù)對12取模后余數(shù)為8，其它數(shù)據(jù)集對其id除以100取整，從而使得id大于9.4億的數(shù)據(jù)在Shuffle時可被均勻分配到所有Task中，而id小于9.4億的數(shù)據(jù)全部分配到同一個Task中。案例18調(diào)整并行度分散同一個Task的不同Key處理過程如下：案例INSERTOVERWRITETABLEtestSELECTCASEWHENid<940000000THEN(9500000+(CAST(RAND()*8ASINTEGER))*12)ELSECAST(id/100ASINTEGER)END,NameFROMstudent_externalWHEREidBETWEEN900000000AND1050000000;通過上述處理，一份可能造成后續(xù)數(shù)據(jù)傾斜的測試數(shù)據(jù)已經(jīng)準(zhǔn)備好。19調(diào)整并行度分散同一個Task的不同Key接下來，使用Spark讀取該測試數(shù)據(jù)，并通過groupByKey(12)對id分組處理，且Shuffle并行度為12。代碼如下：案例publicclassSparkDataSkew{publicstaticvoidmain(String[]args){SparkSessionsparkSession=SparkSession.builder() .appName("SparkDataSkewTunning") .config("hive.metastore.uris","thrift://hadoop1:9083") .enableHiveSupport() .getOrCreate();Dataset<Row>dataframe=sparkSession.sql("select*fromtest");20調(diào)整并行度分散同一個Task的不同Key案例dataframe.toJavaRDD().mapToPair((Rowrow)->newTuple2<Integer,String>(row.getInt(0),row.getString(1))).groupByKey(12).mapToPair((Tuple2<Integer,Iterable<String>>tuple)->{intid=tuple._1();AtomicIntegeratomicInteger=newAtomicInteger(0);tuple._2().forEach((Stringname)->atomicInteger.incrementAndGet()); returnnewTuple2<Integer,Integer>(id,atomicInteger.get());}).count();sparkSession.stop();sparkSession.close();}}21調(diào)整并行度分散同一個Task的不同Key本次實(shí)驗所使用集群節(jié)點(diǎn)數(shù)為4，每個節(jié)點(diǎn)可被Yarn使用的CPU核數(shù)為16，內(nèi)存為16GB。使用如下方式提交上述應(yīng)用，將啟動4個Executor，每個Executor可使用核數(shù)為12（該配置并非生產(chǎn)環(huán)境下的最優(yōu)配置，僅用于本文實(shí)驗），可用內(nèi)存為12GB。案例spark-submit--queueambari--num-executors4--executor-cores12--executor-memory12g--classcom.jasongj.spark.driver.SparkDataSkew--masteryarn--deploy-modeclientSparkExample-with-dependencies-1.0.jar22調(diào)整并行度分散同一個Task的不同Key運(yùn)行結(jié)果如下圖所示：案例23調(diào)整并行度分散同一個Task的不同KeyGroupByStage的Task狀態(tài)如上圖所示，Task8處理的記錄數(shù)為4500萬，遠(yuǎn)大于（9倍于）其它11個Task處理的500萬記錄。而Task8所耗費(fèi)的時間為38秒，遠(yuǎn)高于其它11個Task的平均時間（16秒）。整個Stage的時間也為38秒，該時間主要由最慢的Task8決定。在這種情況下，可以通過調(diào)整Shuffle并行度，使得原來被分配到同一個Task（即該例中的Task8）的不同Key分配到不同Task，從而降低Task8所需處理的數(shù)據(jù)量，緩解數(shù)據(jù)傾斜。案例24調(diào)整并行度分散同一個Task的不同Key通過groupByKey(48)將Shuffle并行度調(diào)整為48，重新提交到Spark。新的Job的GroupByStage所有Task狀態(tài)如下圖所示。案例25調(diào)整并行度分散同一個Task的不同KeyTask8的4500萬，降低了75%左右，而其耗時從原來Task8的38秒降到了24秒。在這種場景下，調(diào)整并行度，并不意味著一定要增加并行度，也可能是減小并行度。從下圖可見，處理記錄數(shù)最多的Task6所處理的記錄數(shù)約為1045萬，耗時為23秒。處理記錄數(shù)最少的Task1處理的記錄數(shù)約為545萬，耗時12秒。案例26調(diào)整并行度分散同一個Task的不同Key適用場景：大量不同的Key被分配到了相同的Task造成該Task數(shù)據(jù)量過大。解決方案：調(diào)整并行度。一般是增大并行度，但有時如本例減小