柚子快報(bào)邀請碼778899分享：Spark第三課

http://yzkb.51969.com/

1.分區(qū)規(guī)則

1.分區(qū)規(guī)則

shuffle 1.打亂順序 2.重新組合

1.分區(qū)的規(guī)則

默認(rèn)與MapReduce的規(guī)則一致,都是按照哈希值取余進(jìn)行分配. 一個(gè)分區(qū)可以多個(gè)組,一個(gè)組的數(shù)據(jù)必須一個(gè)分區(qū)

2. 分組的分區(qū)導(dǎo)致數(shù)據(jù)傾斜怎么解決?

擴(kuò)容 讓分區(qū)變多修改分區(qū)規(guī)則

3.HashMap擴(kuò)容為什么必須是2的倍數(shù)?

當(dāng)不是2的倍數(shù)時(shí), 好多的位置取不到 比如 為5 01234 123都取不到 必須保證,相關(guān)的位數(shù)全是1,所以必定2的倍數(shù) 2的n次方 所以位運(yùn)算不是什么時(shí)候都能用的

2.轉(zhuǎn)換算子

1.單值轉(zhuǎn)換算子

1.filter過濾器

1.注意

過濾只是將數(shù)據(jù)進(jìn)行校驗(yàn),而不是修改數(shù)據(jù). 結(jié)果為true就保留,false就丟棄

2.代碼

JavaSparkContext sc = new JavaSparkContext("local[*]","filter");

List dataList = Arrays.asList("giao","giao2","zhangsan","lisi");

JavaRDD rdd1 = sc.parallelize(dataList);

//JavaRDD rddFilter1 = rdd1.filter(null);

JavaRDD rddFilter2= rdd1.filter(s->s.substring(0,1).toLowerCase().equals("g"));

//rddFilter1.collect().forEach(System.out::println);

System.out.println("----------------------------");

rddFilter2.collect().forEach(System.out::println);

2.dinstinct

1.原理

分組 通過使用分組取重,相同的話,都是一個(gè)組了,所以Key唯一 應(yīng)該是先分組,然后吧K提出來就好了

2.代碼

JavaSparkContext sc = new JavaSparkContext("local[*]","Distinct");

List dataList = Arrays.asList("giao1","gg1","gg1","gg2","gg2","gg1","gg3","gg1","gg5","gg3");

JavaRDD rdd1 = sc.parallelize(dataList);

JavaRDD rddDistinct = rdd1.distinct();

rddDistinct.collect().forEach(System.out::println);

3.排序

1.介紹

sortby方法需要傳3個(gè)參數(shù) 參數(shù)1 排序規(guī)則 參數(shù)2 升序還是降序(false) 默認(rèn)升序(true) 參數(shù)3 排序的分區(qū)數(shù)量(說明方法底層是靠shuffle實(shí)現(xiàn),所以才有改變分區(qū)的能力)

2.排序規(guī)則

排序規(guī)則,是按照結(jié)果去排序 其實(shí)是用結(jié)果生成一個(gè)K值,通過K值進(jìn)行排序,然后展示 V值 或者說權(quán)值, 按照權(quán)值排序 將Value變成K V

3.代碼

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","SparkSort");

List dataList = Arrays.asList("kunkun","giaogiao","GSD","JJ","chenzhen","Lixiaolong");

JavaRDD rdd1 = sc.parallelize(dataList);

JavaRDD rddSort = rdd1.sortBy(s -> {

switch (s.substring(0, 1).toLowerCase()) {

case "k":

return 5;

case "g":

return 3;

case "j":

return 1;

case "c":

return 2;

case "l":

return 4;

}

return null;

}, false, 3);

rddSort.collect().forEach(System.out::println);

}

2.鍵值對轉(zhuǎn)換算子

1.介紹

1.什么是鍵值對轉(zhuǎn)換算子

如何區(qū)分是鍵值對方法還是單值方法呢? 通過參數(shù)來判斷, 如果參數(shù)是一個(gè)值,就是單值,如果是2個(gè),就是鍵值對

2.元組是不是鍵值對?

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","KVRDD");

List dataList = Arrays.asList(1, 2, 3, 4, 5);

JavaRDD rdd1 = sc.parallelize(dataList);

JavaRDD rddmap = rdd1.map(num -> new Tuple2(num, num));

rddmap.collect().forEach(System.out::println);

}

答案是,不是,因?yàn)檫@個(gè)的返回值,是一個(gè)元組,而元組整體,是一個(gè)單值,所以,是單值 只有返回值 是RDD的時(shí)候,才是鍵值對類型算子

3. 使用Pair轉(zhuǎn)換鍵值對算子

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","RddPair");

List dataList = Arrays.asList(1, 2, 3, 4, 5);

JavaRDD rdd = sc.parallelize(dataList);

JavaPairRDD rddPair = rdd.mapToPair(num -> new Tuple2<>(num, num));

rddPair.collect().forEach(System.out::println);

}

4.直接在獲取時(shí)轉(zhuǎn)換鍵值對

這里使用的是parallelizePairs方法 獲取的是JavaPairRDD

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","KVRDD");

JavaPairRDD rddPair = sc.parallelizePairs(Arrays.asList(

new Tuple2<>("a", 1),

new Tuple2<>("a", 2),

new Tuple2<>("b", 1),

new Tuple2<>("b", 1),

new Tuple2<>("c", 2),

new Tuple2<>("c", 1)

));

rddPair.collect().forEach(System.out::println);

}

5.分組來獲取鍵值對

```java

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","RddPair");

List dataList = Arrays.asList("aa","bb","aa","bb","cc");

JavaRDD rdd = sc.parallelize(dataList);

JavaPairRDD> rddGroup = rdd.groupBy(s->s);

rddGroup.collect().forEach(System.out::println);

}

2.mapValue方法

1.介紹

直接對value進(jìn)行操作,不需要管K 當(dāng)然,也有mapKey方法可以無視Value操作Key

2.代碼演示

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","KVRDD");

JavaPairRDD rddPair = sc.parallelizePairs(Arrays.asList(

new Tuple2<>("a", 1),

new Tuple2<>("a", 2),

new Tuple2<>("b", 1),

new Tuple2<>("b", 1),

new Tuple2<>("c", 2),

new Tuple2<>("c", 1)

));

JavaPairRDD mapV = rddPair.mapValues(num -> num * 2);

mapV.collect().forEach(System.out::println);

}

3.WordCount實(shí)現(xiàn)

iter.spliterator().estimateSize()); spliterator Spliterator（Split Iterator）是Java 8引入的一個(gè)新接口，用于支持并行遍歷和操作數(shù)據(jù)。它是Iterator的擴(kuò)展，可以用于在并行流（Parallel Stream）中對數(shù)據(jù)進(jìn)行劃分和遍歷，從而實(shí)現(xiàn)更高效的并行處理 spliterator()方法是在Iterable接口中定義的一個(gè)默認(rèn)方法，用于生成一個(gè)Spliterator對象，以支持?jǐn)?shù)據(jù)的并行遍歷。它的具體作用是將Iterable中的數(shù)據(jù)轉(zhuǎn)換為一個(gè)可以在并行流中使用的Spliterator對象。

estimateSize

estimateSize()方法是Java中Spliterator接口的一個(gè)方法，用于估算Spliterator所包含的元素?cái)?shù)量的大小。Spliterator是用于支持并行遍歷和操作數(shù)據(jù)的接口，而estimateSize()方法提供了一個(gè)估計(jì)值，用于在處理數(shù)據(jù)時(shí)預(yù)測Spliterator包含的元素?cái)?shù)量。

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","RddPair");

List dataList = Arrays.asList("aa","bb","aa","bb","cc");

JavaRDD rdd = sc.parallelize(dataList);

JavaPairRDD> rddGroup = rdd.groupBy(s->s);

JavaPairRDD wordCount = rddGroup.mapValues(iter -> iter.spliterator().estimateSize());

wordCount.collect().forEach(System.out::println);

}

3.groupby 與groupByKey

1 .代碼

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","G1");

JavaPairRDD rddPair;

rddPair = sc.parallelizePairs(Arrays.asList(

new Tuple2<>("a", 1),

new Tuple2<>("a", 2),

new Tuple2<>("b", 1),

new Tuple2<>("b", 1),

new Tuple2<>("c", 2),

new Tuple2<>("c", 1)

));

JavaPairRDD> rddGroupByKey = rddPair.groupByKey();

JavaPairRDD>> rddGroupBy = rddPair.groupBy(t -> t._1);

rddGroupByKey.collect().forEach(System.out::println);

}

2.分析區(qū)別

1.參數(shù) GroupBy是自選規(guī)則 而GroupByKey是將PairRDD的Key當(dāng)做分組規(guī)則2.結(jié)果 GroupBy是將作為單值去分組,即使RDD是Pair, 而GroupByKey 則是將K V分開 ,將V作為組成員

3.注意

GroupByKey是不能進(jìn)行隨意使用的,底層用的含有shuffle,如果計(jì)算平均值,就不能通過GroupByKey直接進(jìn)行計(jì)算.

4.reduce與reduceByKey

1.介紹

多個(gè)變量進(jìn)行同樣的運(yùn)算規(guī)則 Stream是1.8新特性, 計(jì)算的本質(zhì) 兩兩結(jié)合 reduce

2. 代碼

public static void main(String[] args) {

JavaSparkContext sc = new JavaSparkContext("local[*]","Reduce");

JavaPairRDD rddPair;

rddPair = sc.parallelizePairs(Arrays.asList(

new Tuple2<>("a", 1),

new Tuple2<>("a", 2),

new Tuple2<>("b", 1),

new Tuple2<>("b", 1),

new Tuple2<>("c", 2),

new Tuple2<>("c", 1)

));

rddPair.reduceByKey(Integer::sum).collect().forEach(System.out::println);

}

3.理解

相同Key值的V進(jìn)行運(yùn)算,所以底層是有分組的,所以底層是一定有Shuffle,一定有改變分區(qū)的能力,改變分區(qū)數(shù)量和分區(qū)規(guī)則.

4.與groupByKey區(qū)別

reduceByKey 將相同key的數(shù)量中1的V進(jìn)行兩兩聚合 reduceByKey 相同的key兩兩聚合，在shuffle落盤之前對分區(qū)內(nèi)數(shù)據(jù)進(jìn)行聚合，這樣會減少落盤數(shù)據(jù)量，并不會影響最終結(jié)果(預(yù)聚合) 這就是combine

有錢先整IBM小型機(jī)

Shuffle優(yōu)化 1.花錢 2.調(diào)大緩沖區(qū)（溢出次數(shù)減少） 3.

sortByKey 想比較必須實(shí)現(xiàn)可比較的接口 默認(rèn)排序規(guī)則為升序, 通過K對鍵值對進(jìn)行排序

行動(dòng)算子 通過調(diào)用RDD方法讓Spark的功能行動(dòng)起來 map 是在new

轉(zhuǎn)換算子 得到的是RDD 注意 轉(zhuǎn)換跑不起來 行動(dòng)能跑起來 這句話是錯(cuò)誤的

當(dāng)使用sort時(shí),也是能跑起來的,但是還是轉(zhuǎn)換算子 第一行運(yùn)行占用內(nèi)存,第一個(gè)for 運(yùn)算需要內(nèi)存,但是第一行占用了大量內(nèi)存,所以第一行浪費(fèi)了,這就需要懶加載,所以第一行的執(zhí)行時(shí)機(jī)是在第二個(gè)for運(yùn)行前使用的.

注意map collect 不是懶加載,只是沒人調(diào)用他的job(RDD算子內(nèi)部的代碼) RDD算子外部的代碼都是在Driver端

柚子快報(bào)邀請碼778899分享：Spark第三課

http://yzkb.51969.com/

推薦鏈接