Spark 例子

Spark 例子

最常用的转换操作有两个: map 和 filter ，map(func)是将func应用到所有元素，得到一个新的RDD。filter是将func返回为true的元素过滤出来，组成一个新的RDD。一些比较常用的转换如下：

• map(func) 返回一个新的分布式数据集，将数据源的每一个元素传递给函数 func 映射组成。
• filter(func) 返回一个新的数据集，从数据源中选中一些元素通过函数 func 返回 true。
• flatMap(func) 类似于 map，但是每个输入项能被映射成多个输出项(所以 func 必须返回一个 Seq，而不是单个 item)。
• union(otherDataset) 两个RDD求并集
• intersection(otherDataset) 两个RDD求交集
• groupByKey() 作用于(K,V)的数据集，依据K对值进行归并，返回一个(K, Iterable)
• reduceByKey(func) 作用于(K,V)的数据集，依据K对值使用func进行归约，返回一个(K,V)数据集
• sortByKey([asending]) 返回一个依据K进行排序的数据集

最常用的动作就是reduce，将数据集归约为一个结果。一些比较常用的动作如下：

• reduce(func) 按照func函数对数据集进行归约，func接受两个参数，返回一个结果，须满足结合律和交换律，以便于分布式计算。
• count() 返回数据集的元素个数
• first() 返回第一个元素
• take(n) 以数组形式返回集合的前n个元素
• saveAsTextFile(path) 将数据集保存为文本文件

读写文件

val lines = sc.textFile("file:///path_to_local/file")  
val lines = sc.textFile("hdfs:///path_to_hdfs/file")  
rdd.saveAsTextFile("hdfs://")

如果是parquet格式文件，可以用下面的办法，得到一个DataFrame，同样可以识别本地及hdfs文件，也可以识别目录及正则

val parquetFile = sqlContext.read.parquet("people.parquet")  
df.write.save("temp.parquet")  

JSON格式文件

val df = sqlContext.read.json("path to json file")  
val df = sqlContext.read.format("json").load("path to file")  
df.write.format("json").save("path to save")

统计字符数

val lines = sc.textFile("data.txt")     //读文件，得到以行字符串为单位的RDD  
val lineLengths = lines.map(s => s.length)    //转换，将字符串元素映射为其长度   
val totalLength = lineLengths.reduce((a, b) => a + b)   //动作，将所有元素加起来 

---

import org.apache.spark.SparkConf  
import org.apache.spark.SparkContext  
import org.apache.spark.SparkContext._  
  
object SparkWordCount {  
 def FILE_NAME:String = "word_count_results_";  
   
 def main(args:Array[String]) {  
 if (args.length < 1) {  
 println("Usage:SparkWordCount FileName");  
 System.exit(1);  
 }  
 val conf = new SparkConf().setAppName("Spark Exercise: Spark Version Word Count Program");  
 val sc = new SparkContext(conf);  
 val textFile = sc.textFile(args(0));  
 val wordCounts = textFile.flatMap(line => line.split(" ")).map(  
                                        word => (word, 1)).reduceByKey((a, b) => a + b)  
  
                                          
 wordCounts.saveAsTextFile(FILE_NAME+System.currentTimeMillis());  
 println("Word Count program running results are successfully saved.");  
 }  
}  

---

./spark-submit \  
--class com.ibm.spark.exercise.basic.SparkWordCount \  
--master spark://hadoop036166:7077 \  
--num-executors 3 \  
--driver-memory 6g --executor-memory 2g \  
--executor-cores 2 \  
/home/fams/sparkexercise.jar \  
hdfs://hadoop036166:9000/user/fams/*.txt  

求平均值

import org.apache.spark.SparkConf  
import org.apache.spark.SparkContext  
object AvgAgeCalculator {  
 def main(args:Array[String]) {  
 if (args.length < 1){  
 println("Usage:AvgAgeCalculator datafile")  
 System.exit(1)  
 }  
 val conf = new SparkConf().setAppName("Spark Exercise:Average Age Calculator")  
 val sc = new SparkContext(conf)  
 val dataFile = sc.textFile(args(0), 5);  
 val count = dataFile.count()  
 val ageData = dataFile.map(line => line.split(" ")(1))  
 val totalAge = ageData.map(age => Integer.parseInt(  
                                String.valueOf(age))).collect().reduce((a,b) => a+b)  
 println("Total Age:" + totalAge + ";Number of People:" + count )  
 val avgAge : Double = totalAge.toDouble / count.toDouble  
 println("Average Age is " + avgAge)  
 }  
}  

---

./spark-submit \  
 --class com.ibm.spark.exercise.basic.AvgAgeCalculator \  
 --master spark://hadoop036166:7077 \  
 --num-executors 3 \  
 --driver-memory 6g \  
 --executor-memory 2g \  
 --executor-cores 2 \  
 /home/fams/sparkexercise.jar \  
 hdfs://hadoop036166:9000/user/fams/inputfiles/sample_age_data.txt  

求男性/女性 最高 最低身高

object PeopleInfoCalculator {  
 def main(args:Array[String]) {  
 if (args.length < 1){  
 println("Usage:PeopleInfoCalculator datafile")  
 System.exit(1)  
 }  
 val conf = new SparkConf().setAppName("Spark Exercise:People Info(Gender & Height) Calculator")  
 val sc = new SparkContext(conf)  
 val dataFile = sc.textFile(args(0), 5);  
 val maleData = dataFile.filter(line => line.contains("M")).map(  
                              line => (line.split(" ")(1) + " " + line.split(" ")(2)))  
 val femaleData = dataFile.filter(line => line.contains("F")).map(  
                              line => (line.split(" ")(1) + " " + line.split(" ")(2)))  
  
 val maleHeightData = maleData.map(line => line.split(" ")(1).toInt)  
 val femaleHeightData = femaleData.map(line => line.split(" ")(1).toInt)  
  
 val lowestMale = maleHeightData.sortBy(x => x,true).first()  
 val lowestFemale = femaleHeightData.sortBy(x => x,true).first()  
  
 val highestMale = maleHeightData.sortBy(x => x, false).first()  
 val highestFemale = femaleHeightData.sortBy(x => x, false).first()  
 println("Number of Male Peole:" + maleData.count())  
 println("Number of Female Peole:" + femaleData.count())  
 println("Lowest Male:" + lowestMale)  
 println("Lowest Female:" + lowestFemale)  
 println("Highest Male:" + highestMale)  
 println("Highest Female:" + highestFemale)  
 }  
}  

./spark-submit \  
 --class com.ibm.spark.exercise.basic.PeopleInfoCalculator \  
 --master spark://hadoop036166:7077 \  
 --num-executors 3 \  
 --driver-memory 6g \  
 --executor-memory 3g \  
 --executor-cores 2 \  
 /home/fams/sparkexercise.jar \  
 hdfs://hadoop036166:9000/user/fams/inputfiles/sample_people_info.txt  

每行数据出现的次数最高的

=============

import org.apache.spark.SparkConf  
import org.apache.spark.SparkContext  
  
object TopKSearchKeyWords {  
 def main(args:Array[String]){  
 if (args.length < 2) {  
 println("Usage:TopKSearchKeyWords KeyWordsFile K");  
 System.exit(1)  
 }  
 val conf = new SparkConf().setAppName("Spark Exercise:Top K Searching Key Words")  
 val sc = new SparkContext(conf)  
 val srcData = sc.textFile(args(0))  
 val countedData = srcData.map(line => (line.toLowerCase(),1)).reduceByKey((a,b) => a+b)  
  
 val sortedData = countedData.map{ case (k,v) => (v,k) }.sortByKey(false)  
 val topKData = sortedData.take(args(1).toInt).map{ case (v,k) => (k,v) }  
 topKData.foreach(println)  
 }  
}