RDD及其特點

1、RDD是Spark的核心數據模型，但是個抽象類，全稱為Resillient Distributed Dataset，即彈性分佈式數據集。

2、RDD在抽象上來說是一種元素集合，包含了數據。它是被分區的，分為多個分區，每個分區分佈在集群中的不同節點上，從而讓RDD中的數據可以被並行操作。（分佈式數據集）

3、RDD通常通過Hadoop上的文件，即HDFS文件或者Hive表，來進行創建；有時也可以通過應用程序中的集合來創建。

4、RDD最重要的特性就是，提供了容錯性，可以自動從節點失敗中恢復過來。即如果某個節點上的RDDpartition，因為節點故障，導致數據丟了，那麼RDD會自動通過自己的數據來源重新計算該partition。這一切對使用者是透明的。

5、RDD的數據默認情況下存放在內存中的，但是在內存資源不足時，Spark會自動將RDD數據寫入磁盤。（彈性）

創建RDD

進行Spark核心編程的第一步就是創建一個初始的RDD。該RDD，通常就代表和包含了Spark應用程序的輸入源數據。然後通過Spark Core提供的transformation算子，對該RDD進行轉換，來獲取其他的RDD。

Spark Core提供了三種創建RDD的方式：

1.使用程序中的集合創建RDD（主要用於測試）

List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
JavaRDD<Integer> numbersRDD = sc.parallelize(numbers);

2.使用本地文件創建RDD（主要用於臨時性處理有大量數據的文件）

SparkSession spark = SparkSession.builder().master("local").appName("WordCountLocal").getOrCreate();
JavaRDD<String> lines = spark.read().textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt").javaRDD();

3.使用HDFS文件創建RDD（生產環境的常用方式）

SparkSession spark = SparkSession.builder().appName("WordCountCluster").getOrCreate();
JavaRDD<String> lines = spark.read().textFile("hdfs://h0:9000/spark.txt").javaRDD();

使用HDFS文件創建RDD對比使用本地文件創建RDD，需要修改的，只有兩個地方：

第一，將SparkSession對象的master("local")方法去掉

第二，我們針對的不是本地文件了，修改為hadoop hdfs上的真正的存儲大數據的文件

操作RDD

Spark支持兩種RDD操作：transformation和action。

transformation操作

transformation操作會針對已有的RDD創建一個新的RDD。transformation具有lazy特性，即transformation不會觸發spark程序的執行，它們只是記錄了對RDD所做的操作，不會自發的執行。只有執行了一個action，之前的所有transformation才會執行。

常用的transformation介紹：

map ：將RDD中的每個元素傳人自定義函數，獲取一個新的元素，然後用新的元素組成新的RDD。

filter：對RDD中每個元素進行判斷，如果返回true則保留，返回false則剔除。

flatMap：與map類似，但是對每個元素都可以返回一個或多個元素。

groupByKey：根據key進行分組，每個key對應一個Iterable<value>。

reduceByKey：對每個key對應的value進行reduce操作。

sortByKey：對每個key對應的value進行排序操作。

join：對兩個包含<key,value>對的RDD進行join操作，每個keyjoin上的pair，都會傳入自定義函數進行處理。

cogroup：同join，但是每個key對應的Iterable<value>都會傳入自定義函數進行處理。

action操作

action操作主要對RDD進行最後的操作，比如遍歷，reduce，保存到文件等，並可以返回結果給Driver程序。action操作執行，會觸發一個spark job的運行，從而觸發這個action之前所有的transformation的執行，這是action的特性。

常用的action介紹：

reduce：將RDD中的所有元素進行聚合操作。第一個和第二個元素聚合，值與第三個元素聚合，值與第四個元素聚合，以此類推。

collect：將RDD中所有元素獲取到本地客戶端（一般不建議使用）。

count：獲取RDD元素總數。

take(n)：獲取RDD中前n個元素。

saveAsTextFile：將RDD元素保存到文件中，對每個元素調用toString方法。

countByKey：對每個key對應的值進行count計數。

foreach：遍歷RDD中的每個元素。

RDD持久化

要持久化一個RDD，只要調用其cache()或者persist()方法即可。在該RDD第一次被計算出來時，就會直接緩存在每個節點中。但是cache()或者persist()的使用是有規則的，必須在transformation或者textFile等創建了一個RDD之後，直接連續調用cache()或persist()才可以。

如果你先創建一個RDD，然後單獨另起一行執行cache()或persist()方法，是沒有用的，而且會報錯，大量的文件會丟失。

val lines = spark.read.textFile("hdfs://h0:9000/spark.txt").persist()

Spark提供的多種持久化級別，主要是為了在CPU和內存消耗之間進行取捨。

通用的持久化級別的選擇建議：

1、優先使用MEMORY_ONLY，如果可以緩存所有數據的話，那麼就使用這種策略。因為純內存速度最快，而且沒有序列化，不需要消耗CPU進行反序列化操作。

2、如果MEMORY_ONLY策略，無法存儲所有數據的話，那麼使用MEMORY_ONLY_SER，將數據進行序列化進行存儲，純內存操作還是非常快，只是要消耗CPU進行反序列化。

3、如果需要進行快速的失敗恢復，那麼就選擇帶後綴為_2的策略，進行數據的備份，這樣在失敗時，就不需要重新計算了。

4、能不使用DISK相關的策略，就不用使用，有的時候，從磁盤讀取數據，還不如重新計算一次。

共享變量

Spark提供了兩種共享變量：Broadcast Variable（廣播變量）和Accumulator（累加變量）。

BroadcastVariable會將使用到的變量，僅僅為每個節點拷貝一份，更大的用處是優化性能，減少網絡傳輸以及內存消耗。廣播變量是隻讀的。

val factor = 3
val broadcastVars = sc.broadcast(factor);
val numberList = Array(1,2,3,4,5)
val number = sc.parallelize(numberList).map( num => num * broadcastVars.value) //廣播變量讀值broadcastVars.value

Accumulator則可以讓多個task共同操作一份變量，主要可以進行累加操作。task只能對Accumulator進行累加操作，不能讀取它的值。只有Driver程序可以讀取Accumulator的值。

val numberList = Array(1,2,3,4,5)
val numberRDD = sc.parallelize(numberList,1)
val sum = sc.accumulator(0)
numberRDD.foreach{m => sum += m}

小案例實戰1

案例需求：

1、對文本文件內的每個單詞都統計出其出現的次數。

2、按照每個單詞出現次數的數量，降序排序。

步驟：

• 1.創建RDD
• 2.將文本進行拆分 （flatMap)
• 3.將拆分後的單詞進行統計 (mapToPair,reduceByKey)
• 4.反轉鍵值對 (mapToPair)
• 5.按鍵升序排序 (sortedByKey)
• 6.再次反轉鍵值對 (mapToPair)
• 7.打印輸出(foreach)

Java版本jdk1.8以下

public class SortWordCount {
 public static void main(String[] args) throws Exception {
 SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
 JavaSparkContext sc = new JavaSparkContext(conf);
 // 創建lines RDD
 JavaRDD<String> lines = sc.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt");
 // 將文本分割成單詞RDD
 JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
 @Override
 public Iterator<String> call(String s) throws Exception {
 return Arrays.asList(s.split(" ")).iterator();
 }
 });
 //將單詞RDD轉換為（單詞，1）鍵值對RDD
 JavaPairRDD<String,Integer> wordPair = words.mapToPair(new PairFunction<String, String,Integer>() {
 @Override
 public Tuple2<String,Integer> call(String s) throws Exception {
 return new Tuple2<String,Integer>(s,1);
 }
 });
 //對wordPair 進行按鍵計數
 JavaPairRDD<String,Integer> wordCount = wordPair.reduceByKey(new Function2<Integer, Integer, Integer>() {
 @Override
 public Integer call(Integer integer, Integer integer2) throws Exception {
 return integer +integer2;
 }
 });
 // 到這裡為止，就得到了每個單詞出現的次數
 // 我們的新需求，是要按照每個單詞出現次數的順序，降序排序
 // wordCounts RDD內的元素是這種格式：(spark, 3) (hadoop, 2)
 // 因此我們需要將RDD轉換成(3, spark) (2, hadoop)的這種格式，才能根據單詞出現次數進行排序
 
 // 進行key-value的反轉映射
 JavaPairRDD<Integer,String> countWord = wordCount.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
 @Override
 public Tuple2<Integer, String> call(Tuple2<String, Integer> s) throws Exception {
 return new Tuple2<Integer, String>(s._2,s._1);
 }
 });
 // 按照key進行排序
 JavaPairRDD<Integer, String> sortedCountWords = countWord.sortByKey(false);
 // 再次將value-key進行反轉映射
 JavaPairRDD<String,Integer> sortedWordCount = sortedCountWords.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
 @Override
 public Tuple2<String, Integer> call(Tuple2<Integer, String> s) throws Exception {
 return new Tuple2<String, Integer>(s._2,s._1);
 }
 });
 // 到此為止，我們獲得了按照單詞出現次數排序後的單詞計數
 // 打印出來
 sortedWordCount.foreach(new VoidFunction<Tuple2<String, Integer>>() {
 @Override
 public void call(Tuple2<String, Integer> s) throws Exception {
 System.out.println("word \\""+s._1+"\\" appears "+ s._2+" times.");
 }
 });
 sc.close();
 }
}

Java版本jdk1.8

可以使用lambda表達式，簡化代碼：

public class SortWordCount {
 public static void main(String[] args) throws Exception {
 SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
 JavaSparkContext sc = new JavaSparkContext(conf);
 // 創建lines RDD
 JavaRDD<String> lines = sc.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt");
 JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
 JavaPairRDD<String,Integer> wordPair = words.mapToPair(word -> new Tuple2<>(word,1));
 JavaPairRDD<String,Integer> wordCount = wordPair.reduceByKey((a,b) ->(a+b));
 JavaPairRDD<Integer,String> countWord = wordCount.mapToPair(word -> new Tuple2<>(word._2,word._1));
 JavaPairRDD<Integer,String> sortedCountWord = countWord.sortByKey(false);
 JavaPairRDD<String,Integer> sortedWordCount = sortedCountWord.mapToPair(word -> new Tuple2<>(word._2,word._1));
 sortedWordCount.foreach(s->System.out.println("word \\""+s._1+"\\" appears "+ s._2+" times."));
 sc.close();
 }
}

scala版本

由於spark2 有了統一切入口SparkSession，在這裡就使用了SparkSession。

package cn.spark.study.core
import org.apache.spark.sql.SparkSession
object SortWordCount {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("SortWordCount").master("local").getOrCreate()
 val lines = spark.sparkContext.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt")
 val words = lines.flatMap{line => line.split(" ")}
 val wordCounts = words.map{word => (word,1)}.reduceByKey(_ + _)
 val countWord = wordCounts.map{word =>(word._2,word._1)}
 val sortedCountWord = countWord.sortByKey(false)
 val sortedWordCount = sortedCountWord.map{word => (word._2, word._1)}
 sortedWordCount.foreach(s=>
 {
 println("word \\""+s._1+ "\\" appears "+s._2+" times.")
 })
 spark.stop()
 }
}

小案例實戰2

需求：

1、按照文件中的第一列排序。

2、如果第一列相同，則按照第二列排序。

實現步驟：

• 1、實現自定義的key，要實現Ordered接口和Serializable接口，在key中實現自己對多個列的排序算法
• 2、將包含文本的RDD，映射成key為自定義key，value為文本的JavaPairRDD（map）
• 3、使用sortByKey算子按照自定義的key進行排序（sortByKey）
• 4、再次映射，剔除自定義的key，只保留文本行（map）
• 5、打印輸出（foreach）

這裡主要用scala編寫

class SecondSortKey(val first:Int,val second:Int) extends Ordered[SecondSortKey] with Serializable{
 override def compare(that: SecondSortKey): Int = {
 if(this.first - that.first !=0){
 this.first-that.first
 }else{
 this.second-that.second
 }
 }
}
object SecondSort {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("SecondSort").master("local").getOrCreate()
 val lines = spark.sparkContext.textFile("D:\\\\sort.txt")
 val pairs = lines.map{line => (
 new SecondSortKey(line.split(" ")(0).toInt,line.split(" ")(1).toInt),line
 )}
 val sortedParis = pairs.sortByKey()
 val sortedLines = sortedParis.map(pairs => pairs._2)
 sortedLines.foreach(s => println(s))
 spark.stop()
 }
}

小案例實戰3

需求：

對每個班級內的學生成績，取出前3名。（分組取topn）

實現步驟：

1.創建初始RDD

2.對初始RDD的文本行按空格分割，映射為key-value鍵值對

3.對鍵值對按鍵分組

4.獲取分組後每組前3的成績：

• 4.1 遍歷每組，獲取每組的成績
• 4.2 將一組成績轉換成一個數組緩衝
• 4.3 將數組緩衝按從大到小排序
• 4.4 對排序後的數組緩衝取其前三

5.打印輸出

以下是使用scala實現：

object GroupTop3 {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("GroupTop3").master("local").getOrCreate()
 //創建初始RDD
 val lines = spark.sparkContext.textFile("D:\\\\score.txt")
 //對初始RDD的文本行按空格分割，映射為key-value鍵值對
 val pairs = lines.map(line => (line.split(" ")(0), line.split(" ")(1).toInt))
 //對pairs鍵值對按鍵分組
 val groupedPairs = pairs.groupByKey()
 //獲取分組後每組前3的成績
 val top3Score = groupedPairs.map(classScores => {
 var className = classScores._1
 //獲取每組的成績，將其轉換成一個數組緩衝,並按從大到小排序，取其前三
 var top3 = classScores._2.toBuffer.sortWith(_>_).take(3)
 Tuple2(className,top3)
 })
 top3Score.foreach(m => {
 println(m._1)
 for(s <- m._2) println(s)
 println("------------------")
 })
 }
}

以上三個小案例都用Scala實現了，用到了Scala中的集合的操作、高階函數、鏈式調用、隱式轉換等知識，自己動手實現，對Scala有個比較好的理解和掌握。

大家一定要學好大數據，為人工智能做準備，好好的提升一下自己的大數據知識儲備，厚積薄發。

小編這裡正好有一套大數據從入門到精通的視頻教程，希望大家能夠喜歡，，，，，

轉發關注小編，直接私信小編“學習”免費來進行獲取~~~~

"

RDD及其特點

1、RDD是Spark的核心數據模型，但是個抽象類，全稱為Resillient Distributed Dataset，即彈性分佈式數據集。

2、RDD在抽象上來說是一種元素集合，包含了數據。它是被分區的，分為多個分區，每個分區分佈在集群中的不同節點上，從而讓RDD中的數據可以被並行操作。（分佈式數據集）

3、RDD通常通過Hadoop上的文件，即HDFS文件或者Hive表，來進行創建；有時也可以通過應用程序中的集合來創建。

4、RDD最重要的特性就是，提供了容錯性，可以自動從節點失敗中恢復過來。即如果某個節點上的RDDpartition，因為節點故障，導致數據丟了，那麼RDD會自動通過自己的數據來源重新計算該partition。這一切對使用者是透明的。

5、RDD的數據默認情況下存放在內存中的，但是在內存資源不足時，Spark會自動將RDD數據寫入磁盤。（彈性）

創建RDD

進行Spark核心編程的第一步就是創建一個初始的RDD。該RDD，通常就代表和包含了Spark應用程序的輸入源數據。然後通過Spark Core提供的transformation算子，對該RDD進行轉換，來獲取其他的RDD。

Spark Core提供了三種創建RDD的方式：

1.使用程序中的集合創建RDD（主要用於測試）

List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
JavaRDD<Integer> numbersRDD = sc.parallelize(numbers);

2.使用本地文件創建RDD（主要用於臨時性處理有大量數據的文件）

SparkSession spark = SparkSession.builder().master("local").appName("WordCountLocal").getOrCreate();
JavaRDD<String> lines = spark.read().textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt").javaRDD();

3.使用HDFS文件創建RDD（生產環境的常用方式）

SparkSession spark = SparkSession.builder().appName("WordCountCluster").getOrCreate();
JavaRDD<String> lines = spark.read().textFile("hdfs://h0:9000/spark.txt").javaRDD();

使用HDFS文件創建RDD對比使用本地文件創建RDD，需要修改的，只有兩個地方：

第一，將SparkSession對象的master("local")方法去掉

第二，我們針對的不是本地文件了，修改為hadoop hdfs上的真正的存儲大數據的文件

操作RDD

Spark支持兩種RDD操作：transformation和action。

transformation操作

transformation操作會針對已有的RDD創建一個新的RDD。transformation具有lazy特性，即transformation不會觸發spark程序的執行，它們只是記錄了對RDD所做的操作，不會自發的執行。只有執行了一個action，之前的所有transformation才會執行。

常用的transformation介紹：

map ：將RDD中的每個元素傳人自定義函數，獲取一個新的元素，然後用新的元素組成新的RDD。

filter：對RDD中每個元素進行判斷，如果返回true則保留，返回false則剔除。

flatMap：與map類似，但是對每個元素都可以返回一個或多個元素。

groupByKey：根據key進行分組，每個key對應一個Iterable<value>。

reduceByKey：對每個key對應的value進行reduce操作。

sortByKey：對每個key對應的value進行排序操作。

join：對兩個包含<key,value>對的RDD進行join操作，每個keyjoin上的pair，都會傳入自定義函數進行處理。

cogroup：同join，但是每個key對應的Iterable<value>都會傳入自定義函數進行處理。

action操作

action操作主要對RDD進行最後的操作，比如遍歷，reduce，保存到文件等，並可以返回結果給Driver程序。action操作執行，會觸發一個spark job的運行，從而觸發這個action之前所有的transformation的執行，這是action的特性。

常用的action介紹：

reduce：將RDD中的所有元素進行聚合操作。第一個和第二個元素聚合，值與第三個元素聚合，值與第四個元素聚合，以此類推。

collect：將RDD中所有元素獲取到本地客戶端（一般不建議使用）。

count：獲取RDD元素總數。

take(n)：獲取RDD中前n個元素。

saveAsTextFile：將RDD元素保存到文件中，對每個元素調用toString方法。

countByKey：對每個key對應的值進行count計數。

foreach：遍歷RDD中的每個元素。

RDD持久化

要持久化一個RDD，只要調用其cache()或者persist()方法即可。在該RDD第一次被計算出來時，就會直接緩存在每個節點中。但是cache()或者persist()的使用是有規則的，必須在transformation或者textFile等創建了一個RDD之後，直接連續調用cache()或persist()才可以。

如果你先創建一個RDD，然後單獨另起一行執行cache()或persist()方法，是沒有用的，而且會報錯，大量的文件會丟失。

val lines = spark.read.textFile("hdfs://h0:9000/spark.txt").persist()

Spark提供的多種持久化級別，主要是為了在CPU和內存消耗之間進行取捨。

通用的持久化級別的選擇建議：

1、優先使用MEMORY_ONLY，如果可以緩存所有數據的話，那麼就使用這種策略。因為純內存速度最快，而且沒有序列化，不需要消耗CPU進行反序列化操作。

2、如果MEMORY_ONLY策略，無法存儲所有數據的話，那麼使用MEMORY_ONLY_SER，將數據進行序列化進行存儲，純內存操作還是非常快，只是要消耗CPU進行反序列化。

3、如果需要進行快速的失敗恢復，那麼就選擇帶後綴為_2的策略，進行數據的備份，這樣在失敗時，就不需要重新計算了。

4、能不使用DISK相關的策略，就不用使用，有的時候，從磁盤讀取數據，還不如重新計算一次。

共享變量

Spark提供了兩種共享變量：Broadcast Variable（廣播變量）和Accumulator（累加變量）。

BroadcastVariable會將使用到的變量，僅僅為每個節點拷貝一份，更大的用處是優化性能，減少網絡傳輸以及內存消耗。廣播變量是隻讀的。

val factor = 3
val broadcastVars = sc.broadcast(factor);
val numberList = Array(1,2,3,4,5)
val number = sc.parallelize(numberList).map( num => num * broadcastVars.value) //廣播變量讀值broadcastVars.value

Accumulator則可以讓多個task共同操作一份變量，主要可以進行累加操作。task只能對Accumulator進行累加操作，不能讀取它的值。只有Driver程序可以讀取Accumulator的值。

val numberList = Array(1,2,3,4,5)
val numberRDD = sc.parallelize(numberList,1)
val sum = sc.accumulator(0)
numberRDD.foreach{m => sum += m}

小案例實戰1

案例需求：

1、對文本文件內的每個單詞都統計出其出現的次數。

2、按照每個單詞出現次數的數量，降序排序。

步驟：

• 1.創建RDD
• 2.將文本進行拆分 （flatMap)
• 3.將拆分後的單詞進行統計 (mapToPair,reduceByKey)
• 4.反轉鍵值對 (mapToPair)
• 5.按鍵升序排序 (sortedByKey)
• 6.再次反轉鍵值對 (mapToPair)
• 7.打印輸出(foreach)

Java版本jdk1.8以下

public class SortWordCount {
 public static void main(String[] args) throws Exception {
 SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
 JavaSparkContext sc = new JavaSparkContext(conf);
 // 創建lines RDD
 JavaRDD<String> lines = sc.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt");
 // 將文本分割成單詞RDD
 JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
 @Override
 public Iterator<String> call(String s) throws Exception {
 return Arrays.asList(s.split(" ")).iterator();
 }
 });
 //將單詞RDD轉換為（單詞，1）鍵值對RDD
 JavaPairRDD<String,Integer> wordPair = words.mapToPair(new PairFunction<String, String,Integer>() {
 @Override
 public Tuple2<String,Integer> call(String s) throws Exception {
 return new Tuple2<String,Integer>(s,1);
 }
 });
 //對wordPair 進行按鍵計數
 JavaPairRDD<String,Integer> wordCount = wordPair.reduceByKey(new Function2<Integer, Integer, Integer>() {
 @Override
 public Integer call(Integer integer, Integer integer2) throws Exception {
 return integer +integer2;
 }
 });
 // 到這裡為止，就得到了每個單詞出現的次數
 // 我們的新需求，是要按照每個單詞出現次數的順序，降序排序
 // wordCounts RDD內的元素是這種格式：(spark, 3) (hadoop, 2)
 // 因此我們需要將RDD轉換成(3, spark) (2, hadoop)的這種格式，才能根據單詞出現次數進行排序
 
 // 進行key-value的反轉映射
 JavaPairRDD<Integer,String> countWord = wordCount.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
 @Override
 public Tuple2<Integer, String> call(Tuple2<String, Integer> s) throws Exception {
 return new Tuple2<Integer, String>(s._2,s._1);
 }
 });
 // 按照key進行排序
 JavaPairRDD<Integer, String> sortedCountWords = countWord.sortByKey(false);
 // 再次將value-key進行反轉映射
 JavaPairRDD<String,Integer> sortedWordCount = sortedCountWords.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
 @Override
 public Tuple2<String, Integer> call(Tuple2<Integer, String> s) throws Exception {
 return new Tuple2<String, Integer>(s._2,s._1);
 }
 });
 // 到此為止，我們獲得了按照單詞出現次數排序後的單詞計數
 // 打印出來
 sortedWordCount.foreach(new VoidFunction<Tuple2<String, Integer>>() {
 @Override
 public void call(Tuple2<String, Integer> s) throws Exception {
 System.out.println("word \\""+s._1+"\\" appears "+ s._2+" times.");
 }
 });
 sc.close();
 }
}

Java版本jdk1.8

可以使用lambda表達式，簡化代碼：

public class SortWordCount {
 public static void main(String[] args) throws Exception {
 SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
 JavaSparkContext sc = new JavaSparkContext(conf);
 // 創建lines RDD
 JavaRDD<String> lines = sc.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt");
 JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
 JavaPairRDD<String,Integer> wordPair = words.mapToPair(word -> new Tuple2<>(word,1));
 JavaPairRDD<String,Integer> wordCount = wordPair.reduceByKey((a,b) ->(a+b));
 JavaPairRDD<Integer,String> countWord = wordCount.mapToPair(word -> new Tuple2<>(word._2,word._1));
 JavaPairRDD<Integer,String> sortedCountWord = countWord.sortByKey(false);
 JavaPairRDD<String,Integer> sortedWordCount = sortedCountWord.mapToPair(word -> new Tuple2<>(word._2,word._1));
 sortedWordCount.foreach(s->System.out.println("word \\""+s._1+"\\" appears "+ s._2+" times."));
 sc.close();
 }
}

scala版本

由於spark2 有了統一切入口SparkSession，在這裡就使用了SparkSession。

package cn.spark.study.core
import org.apache.spark.sql.SparkSession
object SortWordCount {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("SortWordCount").master("local").getOrCreate()
 val lines = spark.sparkContext.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt")
 val words = lines.flatMap{line => line.split(" ")}
 val wordCounts = words.map{word => (word,1)}.reduceByKey(_ + _)
 val countWord = wordCounts.map{word =>(word._2,word._1)}
 val sortedCountWord = countWord.sortByKey(false)
 val sortedWordCount = sortedCountWord.map{word => (word._2, word._1)}
 sortedWordCount.foreach(s=>
 {
 println("word \\""+s._1+ "\\" appears "+s._2+" times.")
 })
 spark.stop()
 }
}

小案例實戰2

需求：

1、按照文件中的第一列排序。

2、如果第一列相同，則按照第二列排序。

實現步驟：

• 1、實現自定義的key，要實現Ordered接口和Serializable接口，在key中實現自己對多個列的排序算法
• 2、將包含文本的RDD，映射成key為自定義key，value為文本的JavaPairRDD（map）
• 3、使用sortByKey算子按照自定義的key進行排序（sortByKey）
• 4、再次映射，剔除自定義的key，只保留文本行（map）
• 5、打印輸出（foreach）

這裡主要用scala編寫

class SecondSortKey(val first:Int,val second:Int) extends Ordered[SecondSortKey] with Serializable{
 override def compare(that: SecondSortKey): Int = {
 if(this.first - that.first !=0){
 this.first-that.first
 }else{
 this.second-that.second
 }
 }
}
object SecondSort {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("SecondSort").master("local").getOrCreate()
 val lines = spark.sparkContext.textFile("D:\\\\sort.txt")
 val pairs = lines.map{line => (
 new SecondSortKey(line.split(" ")(0).toInt,line.split(" ")(1).toInt),line
 )}
 val sortedParis = pairs.sortByKey()
 val sortedLines = sortedParis.map(pairs => pairs._2)
 sortedLines.foreach(s => println(s))
 spark.stop()
 }
}

小案例實戰3

需求：

對每個班級內的學生成績，取出前3名。（分組取topn）

實現步驟：

1.創建初始RDD

2.對初始RDD的文本行按空格分割，映射為key-value鍵值對

3.對鍵值對按鍵分組

4.獲取分組後每組前3的成績：

• 4.1 遍歷每組，獲取每組的成績
• 4.2 將一組成績轉換成一個數組緩衝
• 4.3 將數組緩衝按從大到小排序
• 4.4 對排序後的數組緩衝取其前三

5.打印輸出

以下是使用scala實現：

object GroupTop3 {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("GroupTop3").master("local").getOrCreate()
 //創建初始RDD
 val lines = spark.sparkContext.textFile("D:\\\\score.txt")
 //對初始RDD的文本行按空格分割，映射為key-value鍵值對
 val pairs = lines.map(line => (line.split(" ")(0), line.split(" ")(1).toInt))
 //對pairs鍵值對按鍵分組
 val groupedPairs = pairs.groupByKey()
 //獲取分組後每組前3的成績
 val top3Score = groupedPairs.map(classScores => {
 var className = classScores._1
 //獲取每組的成績，將其轉換成一個數組緩衝,並按從大到小排序，取其前三
 var top3 = classScores._2.toBuffer.sortWith(_>_).take(3)
 Tuple2(className,top3)
 })
 top3Score.foreach(m => {
 println(m._1)
 for(s <- m._2) println(s)
 println("------------------")
 })
 }
}

以上三個小案例都用Scala實現了，用到了Scala中的集合的操作、高階函數、鏈式調用、隱式轉換等知識，自己動手實現，對Scala有個比較好的理解和掌握。

大家一定要學好大數據，為人工智能做準備，好好的提升一下自己的大數據知識儲備，厚積薄發。

小編這裡正好有一套大數據從入門到精通的視頻教程，希望大家能夠喜歡，，，，，

轉發關注小編，直接私信小編“學習”免費來進行獲取~~~~

"

RDD及其特點

1、RDD是Spark的核心數據模型，但是個抽象類，全稱為Resillient Distributed Dataset，即彈性分佈式數據集。

2、RDD在抽象上來說是一種元素集合，包含了數據。它是被分區的，分為多個分區，每個分區分佈在集群中的不同節點上，從而讓RDD中的數據可以被並行操作。（分佈式數據集）

3、RDD通常通過Hadoop上的文件，即HDFS文件或者Hive表，來進行創建；有時也可以通過應用程序中的集合來創建。

4、RDD最重要的特性就是，提供了容錯性，可以自動從節點失敗中恢復過來。即如果某個節點上的RDDpartition，因為節點故障，導致數據丟了，那麼RDD會自動通過自己的數據來源重新計算該partition。這一切對使用者是透明的。

5、RDD的數據默認情況下存放在內存中的，但是在內存資源不足時，Spark會自動將RDD數據寫入磁盤。（彈性）

創建RDD

進行Spark核心編程的第一步就是創建一個初始的RDD。該RDD，通常就代表和包含了Spark應用程序的輸入源數據。然後通過Spark Core提供的transformation算子，對該RDD進行轉換，來獲取其他的RDD。

Spark Core提供了三種創建RDD的方式：

1.使用程序中的集合創建RDD（主要用於測試）

List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
JavaRDD<Integer> numbersRDD = sc.parallelize(numbers);

2.使用本地文件創建RDD（主要用於臨時性處理有大量數據的文件）

SparkSession spark = SparkSession.builder().master("local").appName("WordCountLocal").getOrCreate();
JavaRDD<String> lines = spark.read().textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt").javaRDD();

3.使用HDFS文件創建RDD（生產環境的常用方式）

SparkSession spark = SparkSession.builder().appName("WordCountCluster").getOrCreate();
JavaRDD<String> lines = spark.read().textFile("hdfs://h0:9000/spark.txt").javaRDD();

使用HDFS文件創建RDD對比使用本地文件創建RDD，需要修改的，只有兩個地方：

第一，將SparkSession對象的master("local")方法去掉

第二，我們針對的不是本地文件了，修改為hadoop hdfs上的真正的存儲大數據的文件

操作RDD

Spark支持兩種RDD操作：transformation和action。

transformation操作

transformation操作會針對已有的RDD創建一個新的RDD。transformation具有lazy特性，即transformation不會觸發spark程序的執行，它們只是記錄了對RDD所做的操作，不會自發的執行。只有執行了一個action，之前的所有transformation才會執行。

常用的transformation介紹：

map ：將RDD中的每個元素傳人自定義函數，獲取一個新的元素，然後用新的元素組成新的RDD。

filter：對RDD中每個元素進行判斷，如果返回true則保留，返回false則剔除。

flatMap：與map類似，但是對每個元素都可以返回一個或多個元素。

groupByKey：根據key進行分組，每個key對應一個Iterable<value>。

reduceByKey：對每個key對應的value進行reduce操作。

sortByKey：對每個key對應的value進行排序操作。

join：對兩個包含<key,value>對的RDD進行join操作，每個keyjoin上的pair，都會傳入自定義函數進行處理。

cogroup：同join，但是每個key對應的Iterable<value>都會傳入自定義函數進行處理。

action操作

action操作主要對RDD進行最後的操作，比如遍歷，reduce，保存到文件等，並可以返回結果給Driver程序。action操作執行，會觸發一個spark job的運行，從而觸發這個action之前所有的transformation的執行，這是action的特性。

常用的action介紹：

reduce：將RDD中的所有元素進行聚合操作。第一個和第二個元素聚合，值與第三個元素聚合，值與第四個元素聚合，以此類推。

collect：將RDD中所有元素獲取到本地客戶端（一般不建議使用）。

count：獲取RDD元素總數。

take(n)：獲取RDD中前n個元素。

saveAsTextFile：將RDD元素保存到文件中，對每個元素調用toString方法。

countByKey：對每個key對應的值進行count計數。

foreach：遍歷RDD中的每個元素。

RDD持久化

要持久化一個RDD，只要調用其cache()或者persist()方法即可。在該RDD第一次被計算出來時，就會直接緩存在每個節點中。但是cache()或者persist()的使用是有規則的，必須在transformation或者textFile等創建了一個RDD之後，直接連續調用cache()或persist()才可以。

如果你先創建一個RDD，然後單獨另起一行執行cache()或persist()方法，是沒有用的，而且會報錯，大量的文件會丟失。

val lines = spark.read.textFile("hdfs://h0:9000/spark.txt").persist()

Spark提供的多種持久化級別，主要是為了在CPU和內存消耗之間進行取捨。

通用的持久化級別的選擇建議：

1、優先使用MEMORY_ONLY，如果可以緩存所有數據的話，那麼就使用這種策略。因為純內存速度最快，而且沒有序列化，不需要消耗CPU進行反序列化操作。

2、如果MEMORY_ONLY策略，無法存儲所有數據的話，那麼使用MEMORY_ONLY_SER，將數據進行序列化進行存儲，純內存操作還是非常快，只是要消耗CPU進行反序列化。

3、如果需要進行快速的失敗恢復，那麼就選擇帶後綴為_2的策略，進行數據的備份，這樣在失敗時，就不需要重新計算了。

4、能不使用DISK相關的策略，就不用使用，有的時候，從磁盤讀取數據，還不如重新計算一次。

共享變量

Spark提供了兩種共享變量：Broadcast Variable（廣播變量）和Accumulator（累加變量）。

BroadcastVariable會將使用到的變量，僅僅為每個節點拷貝一份，更大的用處是優化性能，減少網絡傳輸以及內存消耗。廣播變量是隻讀的。

val factor = 3
val broadcastVars = sc.broadcast(factor);
val numberList = Array(1,2,3,4,5)
val number = sc.parallelize(numberList).map( num => num * broadcastVars.value) //廣播變量讀值broadcastVars.value

Accumulator則可以讓多個task共同操作一份變量，主要可以進行累加操作。task只能對Accumulator進行累加操作，不能讀取它的值。只有Driver程序可以讀取Accumulator的值。

val numberList = Array(1,2,3,4,5)
val numberRDD = sc.parallelize(numberList,1)
val sum = sc.accumulator(0)
numberRDD.foreach{m => sum += m}

小案例實戰1

案例需求：

1、對文本文件內的每個單詞都統計出其出現的次數。

2、按照每個單詞出現次數的數量，降序排序。

步驟：

• 1.創建RDD
• 2.將文本進行拆分 （flatMap)
• 3.將拆分後的單詞進行統計 (mapToPair,reduceByKey)
• 4.反轉鍵值對 (mapToPair)
• 5.按鍵升序排序 (sortedByKey)
• 6.再次反轉鍵值對 (mapToPair)
• 7.打印輸出(foreach)

Java版本jdk1.8以下

public class SortWordCount {
 public static void main(String[] args) throws Exception {
 SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
 JavaSparkContext sc = new JavaSparkContext(conf);
 // 創建lines RDD
 JavaRDD<String> lines = sc.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt");
 // 將文本分割成單詞RDD
 JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
 @Override
 public Iterator<String> call(String s) throws Exception {
 return Arrays.asList(s.split(" ")).iterator();
 }
 });
 //將單詞RDD轉換為（單詞，1）鍵值對RDD
 JavaPairRDD<String,Integer> wordPair = words.mapToPair(new PairFunction<String, String,Integer>() {
 @Override
 public Tuple2<String,Integer> call(String s) throws Exception {
 return new Tuple2<String,Integer>(s,1);
 }
 });
 //對wordPair 進行按鍵計數
 JavaPairRDD<String,Integer> wordCount = wordPair.reduceByKey(new Function2<Integer, Integer, Integer>() {
 @Override
 public Integer call(Integer integer, Integer integer2) throws Exception {
 return integer +integer2;
 }
 });
 // 到這裡為止，就得到了每個單詞出現的次數
 // 我們的新需求，是要按照每個單詞出現次數的順序，降序排序
 // wordCounts RDD內的元素是這種格式：(spark, 3) (hadoop, 2)
 // 因此我們需要將RDD轉換成(3, spark) (2, hadoop)的這種格式，才能根據單詞出現次數進行排序
 
 // 進行key-value的反轉映射
 JavaPairRDD<Integer,String> countWord = wordCount.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
 @Override
 public Tuple2<Integer, String> call(Tuple2<String, Integer> s) throws Exception {
 return new Tuple2<Integer, String>(s._2,s._1);
 }
 });
 // 按照key進行排序
 JavaPairRDD<Integer, String> sortedCountWords = countWord.sortByKey(false);
 // 再次將value-key進行反轉映射
 JavaPairRDD<String,Integer> sortedWordCount = sortedCountWords.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
 @Override
 public Tuple2<String, Integer> call(Tuple2<Integer, String> s) throws Exception {
 return new Tuple2<String, Integer>(s._2,s._1);
 }
 });
 // 到此為止，我們獲得了按照單詞出現次數排序後的單詞計數
 // 打印出來
 sortedWordCount.foreach(new VoidFunction<Tuple2<String, Integer>>() {
 @Override
 public void call(Tuple2<String, Integer> s) throws Exception {
 System.out.println("word \\""+s._1+"\\" appears "+ s._2+" times.");
 }
 });
 sc.close();
 }
}

Java版本jdk1.8

可以使用lambda表達式，簡化代碼：

public class SortWordCount {
 public static void main(String[] args) throws Exception {
 SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
 JavaSparkContext sc = new JavaSparkContext(conf);
 // 創建lines RDD
 JavaRDD<String> lines = sc.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt");
 JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
 JavaPairRDD<String,Integer> wordPair = words.mapToPair(word -> new Tuple2<>(word,1));
 JavaPairRDD<String,Integer> wordCount = wordPair.reduceByKey((a,b) ->(a+b));
 JavaPairRDD<Integer,String> countWord = wordCount.mapToPair(word -> new Tuple2<>(word._2,word._1));
 JavaPairRDD<Integer,String> sortedCountWord = countWord.sortByKey(false);
 JavaPairRDD<String,Integer> sortedWordCount = sortedCountWord.mapToPair(word -> new Tuple2<>(word._2,word._1));
 sortedWordCount.foreach(s->System.out.println("word \\""+s._1+"\\" appears "+ s._2+" times."));
 sc.close();
 }
}

scala版本

由於spark2 有了統一切入口SparkSession，在這裡就使用了SparkSession。

package cn.spark.study.core
import org.apache.spark.sql.SparkSession
object SortWordCount {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("SortWordCount").master("local").getOrCreate()
 val lines = spark.sparkContext.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt")
 val words = lines.flatMap{line => line.split(" ")}
 val wordCounts = words.map{word => (word,1)}.reduceByKey(_ + _)
 val countWord = wordCounts.map{word =>(word._2,word._1)}
 val sortedCountWord = countWord.sortByKey(false)
 val sortedWordCount = sortedCountWord.map{word => (word._2, word._1)}
 sortedWordCount.foreach(s=>
 {
 println("word \\""+s._1+ "\\" appears "+s._2+" times.")
 })
 spark.stop()
 }
}

小案例實戰2

需求：

1、按照文件中的第一列排序。

2、如果第一列相同，則按照第二列排序。

實現步驟：

• 1、實現自定義的key，要實現Ordered接口和Serializable接口，在key中實現自己對多個列的排序算法
• 2、將包含文本的RDD，映射成key為自定義key，value為文本的JavaPairRDD（map）
• 3、使用sortByKey算子按照自定義的key進行排序（sortByKey）
• 4、再次映射，剔除自定義的key，只保留文本行（map）
• 5、打印輸出（foreach）

這裡主要用scala編寫

class SecondSortKey(val first:Int,val second:Int) extends Ordered[SecondSortKey] with Serializable{
 override def compare(that: SecondSortKey): Int = {
 if(this.first - that.first !=0){
 this.first-that.first
 }else{
 this.second-that.second
 }
 }
}
object SecondSort {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("SecondSort").master("local").getOrCreate()
 val lines = spark.sparkContext.textFile("D:\\\\sort.txt")
 val pairs = lines.map{line => (
 new SecondSortKey(line.split(" ")(0).toInt,line.split(" ")(1).toInt),line
 )}
 val sortedParis = pairs.sortByKey()
 val sortedLines = sortedParis.map(pairs => pairs._2)
 sortedLines.foreach(s => println(s))
 spark.stop()
 }
}

小案例實戰3

需求：

對每個班級內的學生成績，取出前3名。（分組取topn）

實現步驟：

1.創建初始RDD

2.對初始RDD的文本行按空格分割，映射為key-value鍵值對

3.對鍵值對按鍵分組

4.獲取分組後每組前3的成績：

• 4.1 遍歷每組，獲取每組的成績
• 4.2 將一組成績轉換成一個數組緩衝
• 4.3 將數組緩衝按從大到小排序
• 4.4 對排序後的數組緩衝取其前三

5.打印輸出

以下是使用scala實現：

object GroupTop3 {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("GroupTop3").master("local").getOrCreate()
 //創建初始RDD
 val lines = spark.sparkContext.textFile("D:\\\\score.txt")
 //對初始RDD的文本行按空格分割，映射為key-value鍵值對
 val pairs = lines.map(line => (line.split(" ")(0), line.split(" ")(1).toInt))
 //對pairs鍵值對按鍵分組
 val groupedPairs = pairs.groupByKey()
 //獲取分組後每組前3的成績
 val top3Score = groupedPairs.map(classScores => {
 var className = classScores._1
 //獲取每組的成績，將其轉換成一個數組緩衝,並按從大到小排序，取其前三
 var top3 = classScores._2.toBuffer.sortWith(_>_).take(3)
 Tuple2(className,top3)
 })
 top3Score.foreach(m => {
 println(m._1)
 for(s <- m._2) println(s)
 println("------------------")
 })
 }
}

以上三個小案例都用Scala實現了，用到了Scala中的集合的操作、高階函數、鏈式調用、隱式轉換等知識，自己動手實現，對Scala有個比較好的理解和掌握。

大家一定要學好大數據，為人工智能做準備，好好的提升一下自己的大數據知識儲備，厚積薄發。

小編這裡正好有一套大數據從入門到精通的視頻教程，希望大家能夠喜歡，，，，，

轉發關注小編，直接私信小編“學習”免費來進行獲取~~~~

"

RDD及其特點

1、RDD是Spark的核心數據模型，但是個抽象類，全稱為Resillient Distributed Dataset，即彈性分佈式數據集。

2、RDD在抽象上來說是一種元素集合，包含了數據。它是被分區的，分為多個分區，每個分區分佈在集群中的不同節點上，從而讓RDD中的數據可以被並行操作。（分佈式數據集）

3、RDD通常通過Hadoop上的文件，即HDFS文件或者Hive表，來進行創建；有時也可以通過應用程序中的集合來創建。

4、RDD最重要的特性就是，提供了容錯性，可以自動從節點失敗中恢復過來。即如果某個節點上的RDDpartition，因為節點故障，導致數據丟了，那麼RDD會自動通過自己的數據來源重新計算該partition。這一切對使用者是透明的。

5、RDD的數據默認情況下存放在內存中的，但是在內存資源不足時，Spark會自動將RDD數據寫入磁盤。（彈性）

創建RDD

進行Spark核心編程的第一步就是創建一個初始的RDD。該RDD，通常就代表和包含了Spark應用程序的輸入源數據。然後通過Spark Core提供的transformation算子，對該RDD進行轉換，來獲取其他的RDD。

Spark Core提供了三種創建RDD的方式：

1.使用程序中的集合創建RDD（主要用於測試）

List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
JavaRDD<Integer> numbersRDD = sc.parallelize(numbers);

2.使用本地文件創建RDD（主要用於臨時性處理有大量數據的文件）

SparkSession spark = SparkSession.builder().master("local").appName("WordCountLocal").getOrCreate();
JavaRDD<String> lines = spark.read().textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt").javaRDD();

3.使用HDFS文件創建RDD（生產環境的常用方式）

SparkSession spark = SparkSession.builder().appName("WordCountCluster").getOrCreate();
JavaRDD<String> lines = spark.read().textFile("hdfs://h0:9000/spark.txt").javaRDD();

使用HDFS文件創建RDD對比使用本地文件創建RDD，需要修改的，只有兩個地方：

第一，將SparkSession對象的master("local")方法去掉

第二，我們針對的不是本地文件了，修改為hadoop hdfs上的真正的存儲大數據的文件

操作RDD

Spark支持兩種RDD操作：transformation和action。

transformation操作

transformation操作會針對已有的RDD創建一個新的RDD。transformation具有lazy特性，即transformation不會觸發spark程序的執行，它們只是記錄了對RDD所做的操作，不會自發的執行。只有執行了一個action，之前的所有transformation才會執行。

常用的transformation介紹：

map ：將RDD中的每個元素傳人自定義函數，獲取一個新的元素，然後用新的元素組成新的RDD。

filter：對RDD中每個元素進行判斷，如果返回true則保留，返回false則剔除。

flatMap：與map類似，但是對每個元素都可以返回一個或多個元素。

groupByKey：根據key進行分組，每個key對應一個Iterable<value>。

reduceByKey：對每個key對應的value進行reduce操作。

sortByKey：對每個key對應的value進行排序操作。

join：對兩個包含<key,value>對的RDD進行join操作，每個keyjoin上的pair，都會傳入自定義函數進行處理。

cogroup：同join，但是每個key對應的Iterable<value>都會傳入自定義函數進行處理。

action操作

action操作主要對RDD進行最後的操作，比如遍歷，reduce，保存到文件等，並可以返回結果給Driver程序。action操作執行，會觸發一個spark job的運行，從而觸發這個action之前所有的transformation的執行，這是action的特性。

常用的action介紹：

reduce：將RDD中的所有元素進行聚合操作。第一個和第二個元素聚合，值與第三個元素聚合，值與第四個元素聚合，以此類推。

collect：將RDD中所有元素獲取到本地客戶端（一般不建議使用）。

count：獲取RDD元素總數。

take(n)：獲取RDD中前n個元素。

saveAsTextFile：將RDD元素保存到文件中，對每個元素調用toString方法。

countByKey：對每個key對應的值進行count計數。

foreach：遍歷RDD中的每個元素。

RDD持久化

要持久化一個RDD，只要調用其cache()或者persist()方法即可。在該RDD第一次被計算出來時，就會直接緩存在每個節點中。但是cache()或者persist()的使用是有規則的，必須在transformation或者textFile等創建了一個RDD之後，直接連續調用cache()或persist()才可以。

如果你先創建一個RDD，然後單獨另起一行執行cache()或persist()方法，是沒有用的，而且會報錯，大量的文件會丟失。

val lines = spark.read.textFile("hdfs://h0:9000/spark.txt").persist()

Spark提供的多種持久化級別，主要是為了在CPU和內存消耗之間進行取捨。

通用的持久化級別的選擇建議：

1、優先使用MEMORY_ONLY，如果可以緩存所有數據的話，那麼就使用這種策略。因為純內存速度最快，而且沒有序列化，不需要消耗CPU進行反序列化操作。

2、如果MEMORY_ONLY策略，無法存儲所有數據的話，那麼使用MEMORY_ONLY_SER，將數據進行序列化進行存儲，純內存操作還是非常快，只是要消耗CPU進行反序列化。

3、如果需要進行快速的失敗恢復，那麼就選擇帶後綴為_2的策略，進行數據的備份，這樣在失敗時，就不需要重新計算了。

4、能不使用DISK相關的策略，就不用使用，有的時候，從磁盤讀取數據，還不如重新計算一次。

共享變量

Spark提供了兩種共享變量：Broadcast Variable（廣播變量）和Accumulator（累加變量）。

BroadcastVariable會將使用到的變量，僅僅為每個節點拷貝一份，更大的用處是優化性能，減少網絡傳輸以及內存消耗。廣播變量是隻讀的。

val factor = 3
val broadcastVars = sc.broadcast(factor);
val numberList = Array(1,2,3,4,5)
val number = sc.parallelize(numberList).map( num => num * broadcastVars.value) //廣播變量讀值broadcastVars.value

Accumulator則可以讓多個task共同操作一份變量，主要可以進行累加操作。task只能對Accumulator進行累加操作，不能讀取它的值。只有Driver程序可以讀取Accumulator的值。

val numberList = Array(1,2,3,4,5)
val numberRDD = sc.parallelize(numberList,1)
val sum = sc.accumulator(0)
numberRDD.foreach{m => sum += m}

小案例實戰1

案例需求：

1、對文本文件內的每個單詞都統計出其出現的次數。

2、按照每個單詞出現次數的數量，降序排序。

步驟：

• 1.創建RDD
• 2.將文本進行拆分 （flatMap)
• 3.將拆分後的單詞進行統計 (mapToPair,reduceByKey)
• 4.反轉鍵值對 (mapToPair)
• 5.按鍵升序排序 (sortedByKey)
• 6.再次反轉鍵值對 (mapToPair)
• 7.打印輸出(foreach)

Java版本jdk1.8以下

public class SortWordCount {
 public static void main(String[] args) throws Exception {
 SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
 JavaSparkContext sc = new JavaSparkContext(conf);
 // 創建lines RDD
 JavaRDD<String> lines = sc.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt");
 // 將文本分割成單詞RDD
 JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
 @Override
 public Iterator<String> call(String s) throws Exception {
 return Arrays.asList(s.split(" ")).iterator();
 }
 });
 //將單詞RDD轉換為（單詞，1）鍵值對RDD
 JavaPairRDD<String,Integer> wordPair = words.mapToPair(new PairFunction<String, String,Integer>() {
 @Override
 public Tuple2<String,Integer> call(String s) throws Exception {
 return new Tuple2<String,Integer>(s,1);
 }
 });
 //對wordPair 進行按鍵計數
 JavaPairRDD<String,Integer> wordCount = wordPair.reduceByKey(new Function2<Integer, Integer, Integer>() {
 @Override
 public Integer call(Integer integer, Integer integer2) throws Exception {
 return integer +integer2;
 }
 });
 // 到這裡為止，就得到了每個單詞出現的次數
 // 我們的新需求，是要按照每個單詞出現次數的順序，降序排序
 // wordCounts RDD內的元素是這種格式：(spark, 3) (hadoop, 2)
 // 因此我們需要將RDD轉換成(3, spark) (2, hadoop)的這種格式，才能根據單詞出現次數進行排序
 
 // 進行key-value的反轉映射
 JavaPairRDD<Integer,String> countWord = wordCount.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
 @Override
 public Tuple2<Integer, String> call(Tuple2<String, Integer> s) throws Exception {
 return new Tuple2<Integer, String>(s._2,s._1);
 }
 });
 // 按照key進行排序
 JavaPairRDD<Integer, String> sortedCountWords = countWord.sortByKey(false);
 // 再次將value-key進行反轉映射
 JavaPairRDD<String,Integer> sortedWordCount = sortedCountWords.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
 @Override
 public Tuple2<String, Integer> call(Tuple2<Integer, String> s) throws Exception {
 return new Tuple2<String, Integer>(s._2,s._1);
 }
 });
 // 到此為止，我們獲得了按照單詞出現次數排序後的單詞計數
 // 打印出來
 sortedWordCount.foreach(new VoidFunction<Tuple2<String, Integer>>() {
 @Override
 public void call(Tuple2<String, Integer> s) throws Exception {
 System.out.println("word \\""+s._1+"\\" appears "+ s._2+" times.");
 }
 });
 sc.close();
 }
}

Java版本jdk1.8

可以使用lambda表達式，簡化代碼：

public class SortWordCount {
 public static void main(String[] args) throws Exception {
 SparkConf conf = new SparkConf().setAppName("SortWordCount").setMaster("local");
 JavaSparkContext sc = new JavaSparkContext(conf);
 // 創建lines RDD
 JavaRDD<String> lines = sc.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt");
 JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
 JavaPairRDD<String,Integer> wordPair = words.mapToPair(word -> new Tuple2<>(word,1));
 JavaPairRDD<String,Integer> wordCount = wordPair.reduceByKey((a,b) ->(a+b));
 JavaPairRDD<Integer,String> countWord = wordCount.mapToPair(word -> new Tuple2<>(word._2,word._1));
 JavaPairRDD<Integer,String> sortedCountWord = countWord.sortByKey(false);
 JavaPairRDD<String,Integer> sortedWordCount = sortedCountWord.mapToPair(word -> new Tuple2<>(word._2,word._1));
 sortedWordCount.foreach(s->System.out.println("word \\""+s._1+"\\" appears "+ s._2+" times."));
 sc.close();
 }
}

scala版本

由於spark2 有了統一切入口SparkSession，在這裡就使用了SparkSession。

package cn.spark.study.core
import org.apache.spark.sql.SparkSession
object SortWordCount {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("SortWordCount").master("local").getOrCreate()
 val lines = spark.sparkContext.textFile("D:\\\\Users\\\\Administrator\\\\Desktop\\\\spark.txt")
 val words = lines.flatMap{line => line.split(" ")}
 val wordCounts = words.map{word => (word,1)}.reduceByKey(_ + _)
 val countWord = wordCounts.map{word =>(word._2,word._1)}
 val sortedCountWord = countWord.sortByKey(false)
 val sortedWordCount = sortedCountWord.map{word => (word._2, word._1)}
 sortedWordCount.foreach(s=>
 {
 println("word \\""+s._1+ "\\" appears "+s._2+" times.")
 })
 spark.stop()
 }
}

小案例實戰2

需求：

1、按照文件中的第一列排序。

2、如果第一列相同，則按照第二列排序。

實現步驟：

• 1、實現自定義的key，要實現Ordered接口和Serializable接口，在key中實現自己對多個列的排序算法
• 2、將包含文本的RDD，映射成key為自定義key，value為文本的JavaPairRDD（map）
• 3、使用sortByKey算子按照自定義的key進行排序（sortByKey）
• 4、再次映射，剔除自定義的key，只保留文本行（map）
• 5、打印輸出（foreach）

這裡主要用scala編寫

class SecondSortKey(val first:Int,val second:Int) extends Ordered[SecondSortKey] with Serializable{
 override def compare(that: SecondSortKey): Int = {
 if(this.first - that.first !=0){
 this.first-that.first
 }else{
 this.second-that.second
 }
 }
}
object SecondSort {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("SecondSort").master("local").getOrCreate()
 val lines = spark.sparkContext.textFile("D:\\\\sort.txt")
 val pairs = lines.map{line => (
 new SecondSortKey(line.split(" ")(0).toInt,line.split(" ")(1).toInt),line
 )}
 val sortedParis = pairs.sortByKey()
 val sortedLines = sortedParis.map(pairs => pairs._2)
 sortedLines.foreach(s => println(s))
 spark.stop()
 }
}

小案例實戰3

需求：

對每個班級內的學生成績，取出前3名。（分組取topn）

實現步驟：

1.創建初始RDD

2.對初始RDD的文本行按空格分割，映射為key-value鍵值對

3.對鍵值對按鍵分組

4.獲取分組後每組前3的成績：

• 4.1 遍歷每組，獲取每組的成績
• 4.2 將一組成績轉換成一個數組緩衝
• 4.3 將數組緩衝按從大到小排序
• 4.4 對排序後的數組緩衝取其前三

5.打印輸出

以下是使用scala實現：

object GroupTop3 {
 def main(args: Array[String]): Unit = {
 val spark = SparkSession.builder().appName("GroupTop3").master("local").getOrCreate()
 //創建初始RDD
 val lines = spark.sparkContext.textFile("D:\\\\score.txt")
 //對初始RDD的文本行按空格分割，映射為key-value鍵值對
 val pairs = lines.map(line => (line.split(" ")(0), line.split(" ")(1).toInt))
 //對pairs鍵值對按鍵分組
 val groupedPairs = pairs.groupByKey()
 //獲取分組後每組前3的成績
 val top3Score = groupedPairs.map(classScores => {
 var className = classScores._1
 //獲取每組的成績，將其轉換成一個數組緩衝,並按從大到小排序，取其前三
 var top3 = classScores._2.toBuffer.sortWith(_>_).take(3)
 Tuple2(className,top3)
 })
 top3Score.foreach(m => {
 println(m._1)
 for(s <- m._2) println(s)
 println("------------------")
 })
 }
}

以上三個小案例都用Scala實現了，用到了Scala中的集合的操作、高階函數、鏈式調用、隱式轉換等知識，自己動手實現，對Scala有個比較好的理解和掌握。

大家一定要學好大數據，為人工智能做準備，好好的提升一下自己的大數據知識儲備，厚積薄發。

小編這裡正好有一套大數據從入門到精通的視頻教程，希望大家能夠喜歡，，，，，

轉發關注小編，直接私信小編“學習”免費來進行獲取~~~~

"

相關推薦

'阿里巴巴程序員推出了的一份Linux 學習清單，這一定是你最需要的'

"第一階段Linux+搜索+hadoop體系Linux基礎→shell編程→高併發架構→lucene,solr搜索→hadoop體系→HDFS→mapreduce→hbase→zookeeper→hive→flume→sqoop→項目實戰第二階段機器學習R語言→mahout...

Linux Spark 程序員 HDFS MapReduce Hadoop 大數據 HBase Storm 阿里巴巴集團 Java 工程師 機器學習 算法 雲計算 Lucene Python Docker Hive 好好學習 網絡硬盤 Solr 數據庫 R語言 數據挖掘 2019-09-09

'孩子有這個表現說明以後長得高，你家孩子有嗎？'

"對於家長來說，孩子的身高是非常重要的。一個成年人，如果他的外貌一般，出眾的身高卻可以讓其加分不少。在大部分人的觀念中，影響身高的主要因素的遺傳，確實，如果父母的個子高，那麼自然孩子也不會矮到哪裡去。一般來說，如果是兒子，那麼其身高可以通過父母身高的相加乘1.078再除2得...

不完美媽媽 睡眠 遺傳 體育 牛奶 跑步 雞蛋 大數據 戶外運動 2019-09-03

'學會這招，你也能千人千面，凡見必買'

"​在之前的課程裡我們已經講過該如何啟動營銷，我們用了談對象的例子延伸到如何談客戶。那麼，這裡提出一個問題：你們在追自己喜歡的女孩或男孩的時候，是不是對方喜歡什麼就買什麼送給對方？比如，女生喜歡粉色，你挑禮物時都會首選粉色，女生就很容易接受；或者這個男生比喜歡吃辣，你卻天天...

市場營銷 最in買手君 銀行 天貓 技術 大數據 懷孕 設計 2019-08-31

'清華學霸總結26周Java學習計劃，免費神級教程帶你飛'

"我一直相信：一切不能實實在在幫助到大家的資料，都是耍流氓我的目標是： 做一次又一次真正有價值、愉悅感十足的分享Java+大數據+互聯網架構，不斷深入前沿領域需要本文資料可以關注、私信小編“資料”獲取！！！學習Java，首先要學JavaSE，這是毋庸置疑的。與此同時，和Ja...

Java Eclipse 算法 數據結構 程序員 清華大學 人生第一份工作 設計 編程語言 技術 Apache 設計模式 讀書 集成開發環境 小遊戲 大數據 2019-08-30

'關於智博會的這幾個好消息，你都接收到了嗎？'

"8月27日是智博會對普通觀眾開放的第一天在重慶國際博覽中心現場人山人海 各種炫酷的高科技、花樣百出的互動體驗項目讓大家覺得十分過癮不少人都難得地“刷”了一波步數熱鬧之餘在智博會如火如荼進行的這幾天還傳來不少好消息這些好消息將為重慶經濟社會的發展提供新一輪澎湃動力讓重慶在智...

重慶 我的第一部5G手機 技術 騰訊雲計算 無人駕駛 百度 大數據 騰訊 華為公司 經濟 奉節 投資 阿里巴巴集團 交通 中國汽研 軟件 工程師 聯想集團 大學 電信 2019-08-29

'這個基於Python的戰爭遊戲你是否玩過？它將顛覆你對Python的認知'

"很多人認為Python只適合用來做大數據，或者是爬蟲之類，而對於寫遊戲，他們認為Python實在是個loser，然而這幾個遊戲，你們玩過嗎？第一種遊戲：這是一款基於Python的戰爭遊戲，在國外的名字不亞於國內的某些策略遊戲。第二種遊戲：這是一款非常有意思的3D遊戲：Ga...

Python 戰爭遊戲 網絡爬蟲 小遊戲 大數據 策略遊戲 2019-08-25

'500元的大牌電視見過嗎？拼多多再創紀錄，網友：全村人都能追劇'

"最近，拼多多又開啟了"超低價模式"，而這次它盯上的不是蘋果手機，不是品牌空調，而是大牌智能大屏電視！以前買個大屏電視，動輒幾千，有時還不能保證使用體驗。但是，這次拼多多與兆馳合作，推出的JVC品牌智能大屏電視，可能就要改變這種局面了。首先說這JVC，它是日本第一臺電視機的...

智能電視 JVC iPhone 電視機 良心國貨大賞 技術 設計 索尼 日本 大數據 夏普公司 歷史 農村 文化 2019-08-24

'網貸逾期都是怎麼被催收的？這3點一定會有！你都清楚嗎？'

"辦理過網貸業務的朋友都知道，網貸一旦逾期，就會有催收人員來跟你聯繫，要求借款人儘快償還欠款，並且在催收時，工作人員還會告訴借款人一些逾期的後果。 有不少人會認為催收員是為了嚇唬借款人還款的，雖然很多催收員確實會誇大後果或者是虛構後果，但是催收人員對你說的這些話可能會真的發...

信用記錄關愛日 信用卡 大數據 2019-08-18

'Python不能做遊戲？都是鬼扯，這四個遊戲你都玩過麼？'

"Python是一門功能強大的語言，學好了Python，遊戲、web、爬蟲、大數據等，沒什麼是你不能駕馭的，而且Python經過三十年的起伏發展，到了現在已經是一門非常優秀的語言。很多人認為Python只適合用來做大數據，或者是爬蟲之類，而對於寫遊戲，他們認為Python實...

Python 小遊戲 網絡爬蟲 戰爭遊戲 大數據 策略遊戲 2019-08-10

'“知識”能與實體商業連接嗎？我們去了趟杭州黃龍，尋找這個答案'

"除了“用知識去發現更大的世界”這個好噱頭外，更重要的是看萬科在“科技創新”的探索。這是銳裘第二篇杭州空間考察文。為什麼要關注杭州看商業？除了和上海離得近之外，作為全國前二的新一線城市，杭州還有這兩點一直非常吸引我：其中一點，是依託於自然環境、教育等條件，杭州擁有自上而下的...

杭州 創新科技 上海 文化 智能家居 阿里巴巴集團 易烊千璽代言喜馬拉雅APP 大學 華為公司 蘋果公司 電腦 西安限購升級 瑞安 西湖 今日頭條 交通 人生第一份工作 文章 浙江大學 大數據 2019-08-10

'“不限量”套餐被叫停，你被坑過嗎？'

"不知道大家有沒有用過三大運營商的“不限量”套餐？當初說的好好的流量隨便用，卻悄悄加上了霸王條款——用滿 40GB 就限速。咆哥之前就上過當，辦理了月租好幾百的不限量套餐，以為可以美滋滋地用手機看線上大片，結果沒到半個月就被限速到幾十kb/s ，別說看視頻，甚至聊微信看微博...

運營商 中國電信 我的第一部5G手機 中國聯通 中國移動 電信 移動互聯網 大數據 新京報 2019-08-07

'關於眾創空間的四個要點知識，你都知道嗎？'

"一、發展眾創空間正當時建設“低成本、便利化、全要素”的眾創空間。是對全球創業大爆炸時代的呼應，也是打造我國經濟新引擎的重大舉措。1.中國繁榮動力來源於民眾活力大眾創新是國家繁榮的關鍵。2013年，諾貝爾經濟學獎獲得者（2006）、美國哥倫比亞大學教授埃德蒙•菲爾普斯的《大...

創業 經濟 創客 開源硬件 技術 極客 大學 大眾汽車 文化 物聯網 創新工場 大數據 金融 開源軟件 阿里巴巴集團 諾貝爾經濟學獎 京東商城 硬件 青島 機器人 2019-08-05

'大學“好”專業具有這5個特點，只要佔1個就算，你選對了嗎？'

"2019年高考錄取如火如荼地正在進行著，不出意外的話，本科提前批次與本科一批次的同學應該都收到了屬於自己的大學錄取通知書。那麼，所有已經被錄取的同學或者即將要被錄取的同學，無論我們當時填報志願的時候選擇了多少個專業，也無論我們最喜歡的專業有多少個，我們最終都會被特定的一個...

大學 人工智能 高考 電腦 軟件 臨床醫學 雷軍 金融 並行計算 招聘 大數據 物聯網 2019-08-02

'知識與技術大變革的前夜，風暴正在來襲，你準備好了嗎？'

"如今的貿易戰、中國對於美國、日本對於韓國，本質上都在爭奪技術的至高點，掌控知識產權，從而影響或者控制市場，進而為政治和經濟影響力服務！一、強者恆強，贏者通吃的時代人工智能的大潮，實際上正在掀起另一場技術革命，各個國家都在拼命的投入研發，搶奪市場，其實這個市場並不重要，技術...

技術 知識產權 人工智能 經濟 華為公司 軟件 電腦 操作系統 文化 機器人 ARM 金融 銀行 政治 英特爾 投資 英偉達 大數據 韓國 F-35閃電II戰鬥攻擊機 日本 Windows 美團網 Android 通信 電子商務 能源 農業 2019-07-31

'帶你瞭解國家扶持大數據要學習什麼知識大數據學習的內容有哪些？'

"大數據作為2019年比較熱門的技術，受到越來越多的關注，那麼對於一個想進入大數據的朋友來說，最想知道的是：大數據學什麼？今天科多大數據就和你們一起來分享一篇關於大數據學習內容體系介紹的文章。（資料在尾部）大數據技術體系太龐雜了，基礎技術覆蓋數據採集、數據預處理、分佈式存儲...

大數據 技術 人工智能 圖像處理 數據庫 深度學習 算法 數據挖掘 自然語言處理 電腦 移動互聯網 物聯網 智能手機 分佈式計算 雲計算 Google 可視化技術 並行計算 心理學 機器人 NoSQL 金融 語音識別技術 人生第一份工作 Excel 交通 2019-07-30

'恭喜！泉州這28名學霸，被北大清華錄取！快看有你母校的嗎？'

"最新喜訊！剛從北京傳來一個大！好！消！息！眾所矚目的2019年清華、北大錄取名單出爐！截至目前已知泉州已有28名學霸被北大清華錄取其中21人選擇了北京大學7人選擇了清華大學理 科學校姓名學校泉州七中吳泉霖北大泉州五中葉航北大季延中學羅宇標北大泉州五中林昊葦北大永春一中林芸...

清華大學 北京大學 泉州 晉江 永春縣 數學 哲學 福建 南安 陳興良 歷史學 歷史 政治 大數據 法律 2019-07-28

'《長安十二時辰》稀有名詞及人物原型詳解，這可能是總結最全的了'

"《長安十二時辰》這部劇最近大火，其過人之處就不多加贅述了，相信每個觀看的人都能體會的到。下面要說的，是這部劇裡面的各種稀有名詞及人物原型的詳解，全弄懂後，再看這部劇，你的感受一定會不一樣！先說那些稀有名詞吧！畢竟很多對歷史有濃厚興趣的人，可能之前都沒聽說過。1、口檀：作用...

長安 唐朝 長安十二時辰小課堂 奢侈品 安祿山 張九齡 新唐書 唐肅宗 日本 李白 楊國忠 李泌 賀知章 我在宮裡做廚師 許和子 孟浩然 李林甫 新疆 岑參 華陰 鳳凰 溫暖冬至 楊貴妃 大數據 駱賓王 政治 唐玄宗 2019-07-28

'山西太原的龍城鐵騎你真正瞭解過嗎？他們為城市做出了很大的貢獻'

"今天突然看到一個特別帥氣的視頻宣傳片，是介紹我們太原的一支特殊隊伍，這支隊伍真的是很具有太原的特色，名字就是龍城鐵騎，這真的是，車聲轟鳴，警燈閃爍，龍城鐵騎，震撼亮相看完這個宣傳片，我特意查閱了一下資料，這支隊伍，龍城鐵騎真的是很具有科技感的一支隊伍，首先，鐵騎是在互聯網...

太原 交通 山西 摩托車 大數據 技術 2019-07-23

'拼多多開店解決了這幾個問題，你的店鋪也能月入過萬'

"對於很多賣家來說，每個人都知道標籤的重要性，標籤做的越精準，引來的流量也就越精準，店鋪的銷售額才會越高，這也直接決定了推廣成本和推廣效果，當然你還有不懂的私信我學習 我會不定時回覆你 話不多說請認真閱讀文章。一、標籤形成想要快速形成標籤，直通車是引進精準流量，快速強化標籤...

大數據 收藏 讀書 2019-07-23

'零基礎學習Python元素知識總結（三）入門提升大牛必修'

"零基礎學習Python知識（三）入門提升大牛必修上節我們講到了for語句和用range()來創建元素，咱們在這個基礎上於講一點，值得思考的東西。先來看例子：——————————————numbers = [value**2 for value in range(1,11)...

Python Numbers 數據結構 2019-07-22

推薦中...