知道數據傾斜發生在哪一個stage之后，接著我們就需要根據stage劃分原理，推算出來發生傾斜的那個stage對應代碼中的哪一部分，這部分代碼中肯定會有一個shuffle類算子。精準推算stage與代碼的對應關系，需要對Spark的源碼有深入的理解，這里我們可以介紹一個相對簡單實用的推算方法：只要看到Spark代碼中出現了一個shuffle類算子或者是Spark SQL的SQL語句中出現了會導致shuffle的語句（比如group by語句），那么就可以判定，以那個地方為界限劃分出了前后兩個stage?！ ∵@里我們就以Spark最基礎的入門程序——單詞計數來舉例，如何用最簡單的方法大致推算出一個stage對應的代碼。如下示例，在整個代碼中，只有一個reduceByKey是會發生shuffle的算子，因此就可以認為，以這個算子為界限，會劃分出前后兩個stage?！ ?、stage0，主要是執行從textFile到map操作，以及執行shuffle write操作。shuffle write操作，我們可以簡單理解為對pairs RDD中的數據進行分區操作，每個task處理的數據中，相同的key會寫入同一個磁盤文件內?！ ?、stage1，主要是執行從reduceByKey到collect操作，stage1的各個task一開始運行，就會首先執行shuffle read操作。執行shuffle read操作的task，會從stage0的各個task所在節點拉取屬于自己處理的那些key，然后對同一個key進行全局性的聚合或join等操作，在這里就是對key的value值進行累加。stage1在執行完reduceByKey算子之后，就計算出了最終的WordCounts RDD，然后會執行collect算子，將所有數據拉取到Driver上，供我們遍歷和打印輸出。[python] view plain copy 
val conf = new SparkConf() val sc = new SparkContext(conf) val lines = sc.textFile("hdfs://...") val words = lines.flatMap(_.split(" ")) val pairs = words.map((_, 1)) val wordCounts = pairs.reduceByKey(_ + _) wordCounts.collect().foreach(val sampledPairs = pairs.sample(false, 0.1) val sampledWordCounts = sampledPairs.countByKey() sampledWordCounts.foreach(println(_)) 
[python] view plain copy // 第一步，給RDD中的每個key都打上一個隨機前綴。 JavaPairRDD<String, Long> randomPrefixRdd = rdd.mapToPair( new PairFunction<Tuple2<Long,Long>, String, Long>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Long> call(Tuple2<Long, Long> tuple) throws Exception { Random random = new Random(); int prefix = random.nextInt(10); return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2); } }); // 第二步，對打上隨機前綴的key進行局部聚合。 JavaPairRDD<String, Long> localAggrRdd = randomPrefixRdd.reduceByKey( new Function2<Long, Long, Long>() { private static final long serialVersionUID = 1L; @Override public Long call(Long v1, Long v2) throws Exception { return v1 + v2; } }); // 第三步，去除RDD中每個key的隨機前綴。 JavaPairRDD<Long, Long> removedRandomPrefixRdd = localAggrRdd.mapToPair( new PairFunction<Tuple2<String,Long>, Long, Long>() { private static final long serialVersionUID = 1L; @Override public Tuple2<Long, Long> call(Tuple2<String, Long> tuple) throws Exception { long originalKey = Long.valueOf(tuple._1.split("_")[1]); return new Tuple2<Long, Long>(originalKey, tuple._2); } }); // 第四步，對去除了隨機前綴的RDD進行全局聚合。 JavaPairRDD<Long, Long> globalAggrRdd = removedRandomPrefixRdd.reduceByKey( new Function2<Long, Long, Long>() { private static final long serialVersionUID = 1L; @Override public Long call(Long v1, Long v2) throws Exception { return v1 + v2; } }); // 首先將數據量比較小的RDD的數據，collect到Driver中來。 List<Tuple2<Long, Row>> rdd1Data = rdd1.collect() // 然后使用Spark的廣播功能，將小RDD的數據轉換成廣播變量，這樣每個Executor就只有一份RDD的數據。 // 可以盡可能節省內存空間，并且減少網絡傳輸性能開銷。 final Broadcast<List<Tuple2<Long, Row>>> rdd1DataBroadcast = sc.broadcast(rdd1Data); // 對另外一個RDD執行map類操作，而不再是join類操作。 JavaPairRDD<String, Tuple2<String, Row>> joinedRdd = rdd2.mapToPair( new PairFunction<Tuple2<Long,String>, String, Tuple2<String, Row>>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Tuple2<String, Row>> call(Tuple2<Long, String> tuple) throws Exception { // 在算子函數中，通過廣播變量，獲取到本地Executor中的rdd1數據。 List<Tuple2<Long, Row>> rdd1Data = rdd1DataBroadcast.value(); // 可以將rdd1的數據轉換為一個Map，便于后面進行join操作。 Map<Long, Row> rdd1DataMap = new HashMap<Long, Row>(); for(Tuple2<Long, Row> data : rdd1Data) { rdd1DataMap.put(data._1, data._2); } // 獲取當前RDD數據的key以及value。 String key = tuple._1; String value = tuple._2; // 從rdd1數據Map中，根據key獲取到可以join到的數據。 Row rdd1Value = rdd1DataMap.get(key); return new Tuple2<String, String>(key, new Tuple2<String, Row>(value, rdd1Value)); } }); // 這里得提示一下。 // 上面的做法，僅僅適用于rdd1中的key沒有重復，全部是唯一的場景。 // 如果rdd1中有多個相同的key，那么就得用flatMap類的操作，在進行join的時候不能用map，而是得遍歷rdd1所有數據進行join。 // rdd2中每條數據都可能會返回多條join后的數據。 // 首先從包含了少數幾個導致數據傾斜key的rdd1中，采樣10%的樣本數據。 JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1); // 對樣本數據RDD統計出每個key的出現次數，并按出現次數降序排序。 // 對降序排序后的數據，取出top 1或者top 100的數據，也就是key最多的前n個數據。 // 具體取出多少個數據量最多的key，由大家自己決定，我們這里就取1個作為示范。 JavaPairRDD<Long, Long> mappedSampledRDD = sampledRDD.mapToPair( new PairFunction<Tuple2<Long,String>, Long, Long>() { private static final long serialVersionUID = 1L; @Override public Tuple2<Long, Long> call(Tuple2<Long, String> tuple) throws Exception { return new Tuple2<Long, Long>(tuple._1, 1L); } }); JavaPairRDD<Long, Long> countedSampledRDD = mappedSampledRDD.reduceByKey( new Function2<Long, Long, Long>() { private static final long serialVersionUID = 1L; @Override public Long call(Long v1, Long v2) throws Exception { return v1 + v2; } }); JavaPairRDD<Long, Long> reversedSampledRDD = countedSampledRDD.mapToPair( new PairFunction<Tuple2<Long,Long>, Long, Long>() { private static final long serialVersionUID = 1L; @Override public Tuple2<Long, Long> call(Tuple2<Long, Long> tuple) throws Exception { return new Tuple2<Long, Long>(tuple._2, tuple._1); } }); final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2; // 從rdd1中分拆出導致數據傾斜的key，形成獨立的RDD。 JavaPairRDD<Long, String> skewedRDD = rdd1.filter( new Function<Tuple2<Long,String>, Boolean>() { private static final long serialVersionUID = 1L; @Override public Boolean call(Tuple2<Long, String> tuple) throws Exception { return tuple._1.equals(skewedUserid); } }); // 從rdd1中分拆出不導致數據傾斜的普通key，形成獨立的RDD。 JavaPairRDD<Long, String> commonRDD = rdd1.filter( new Function<Tuple2<Long,String>, Boolean>() { private static final long serialVersionUID = 1L; @Override public Boolean call(Tuple2<Long, String> tuple) throws Exception { return !tuple._1.equals(skewedUserid); } }); // rdd2，就是那個所有key的分布相對較為均勻的rdd。 // 這里將rdd2中，前面獲取到的key對應的數據，過濾出來，分拆成單獨的rdd，并對rdd中的數據使用flatMap算子都擴容100倍。 // 對擴容的每條數據，都打上0～100的前綴。 JavaPairRDD<String, Row> skewedRdd2 = rdd2.filter( new Function<Tuple2<Long,Row>, Boolean>() { private static final long serialVersionUID = 1L; @Override public Boolean call(Tuple2<Long, Row> tuple) throws Exception { return tuple._1.equals(skewedUserid); } }).flatMapToPair(new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() { private static final long serialVersionUID = 1L; @Override public Iterable<Tuple2<String, Row>> call( Tuple2<Long, Row> tuple) throws Exception { Random random = new Random(); List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>(); for(int i = 0; i < 100; i++) { list.add(new Tuple2<String, Row>(i + "_" + tuple._1, tuple._2)); } return list; } }); // 將rdd1中分拆出來的導致傾斜的key的獨立rdd，每條數據都打上100以內的隨機前綴。 // 然后將這個rdd1中分拆出來的獨立rdd，與上面rdd2中分拆出來的獨立rdd，進行join。 JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD1 = skewedRDD.mapToPair( new PairFunction<Tuple2<Long,String>, String, String>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, String> call(Tuple2<Long, String> tuple) throws Exception { Random random = new Random(); int prefix = random.nextInt(100); return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2); } }) .join(skewedUserid2infoRDD) .mapToPair(new PairFunction<Tuple2<String,Tuple2<String,Row>>, Long, Tuple2<String, Row>>() { private static final long serialVersionUID = 1L; @Override public Tuple2<Long, Tuple2<String, Row>> call( Tuple2<String, Tuple2<String, Row>> tuple) throws Exception { long key = Long.valueOf(tuple._1.split("_")[1]); return new Tuple2<Long, Tuple2<String, Row>>(key, tuple._2); } }); // 將rdd1中分拆出來的包含普通key的獨立rdd，直接與rdd2進行join。 JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(rdd2); // 將傾斜key join后的結果與普通key join后的結果，uinon起來。 // 就是最終的join結果。 JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD = joinedRDD1.union(joinedRDD2); // 首先將其中一個key分布相對較為均勻的RDD膨脹100倍。 JavaPairRDD<String, Row> expandedRDD = rdd1.flatMapToPair( new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() { private static final long serialVersionUID = 1L; @Override public Iterable<Tuple2<String, Row>> call(Tuple2<Long, Row> tuple) throws Exception { List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>(); for(int i = 0; i < 100; i++) { list.add(new Tuple2<String, Row>(0 + "_" + tuple._1, tuple._2)); } return list; } }); // 其次，將另一個有數據傾斜key的RDD，每條數據都打上100以內的隨機前綴。 JavaPairRDD<String, String> mappedRDD = rdd2.mapToPair( new PairFunction<Tuple2<Long,String>, String, String>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, String> call(Tuple2<Long, String> tuple) throws Exception { Random random = new Random(); int prefix = random.nextInt(100); return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2); } }); // 將兩個處理后的RDD進行join即可。 JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD); 新聞熱點
疑難解答
