Spark之Scala语言常见应用举例

       作为一个初学者，初次学习Spark，分享一下自己的心得。

        在学习Spark编程时，首先得准备编译环境，确定编程语言，本人用的是Scala语言，IntelliJ IDEA编译环境，同时得准备四个包，分别是：spark-assembly-1.3.1-hd-2.6.0.jar、scala-compiler.jar、scala-library.jar、scala-reflect.jar。将这四个包导入，才能开始自己的Scala编程之旅。

       由于Hadoop环境没有搭建好，所以在练习Scala编程的时候，就不能再Hadoop之上读取HDFS的数据，不过不碍事，为了练习编程，我们可以读取本地的txt文件，然后将结果保存到txt中，这样不仅能感受到Spark RDD的强大，也能达到我们练习编程的目的。下来主要是用实例说明一下Spark RDD常用的一下操作。

       首先我们得配置SparkConf()，一般是读取HDFS上的文件，但是这里读取本地txt文件，配置SparkConf()如下：

<span style="font-size:18px;"><span style="font-size:18px;">conf=new SparkConf().setAppName("Test").setMaster("local[4]")</span></span>
           

        解释一下：Local[N]：本地模式，使用 N 个线程。

        下面这个程序是使用count（）统计行数

<span style="font-size:18px;">object yb {
  /*
   统计行内容出现的次数，即相同行一共出现多少次
   */
  def main(args: Array[String]): Unit={
    val conf=new SparkConf().setAppName("Test").setMaster("local[4]")
    val sc = new SparkContext(conf)
    val lines = sc.textFile("E:/Spark/天池大数据/data_format1/yb.txt")
    val countx=lines.count()//统计行数
    println(countx)//输出： 10485750 
  }
}</span>
           

       统计词频并按照词频排序：

<span style="font-size:18px;">object yb {
  def main(args: Array[String]): Unit={

    val conf=new SparkConf().setAppName("Test").setMaster("local[4]")
    val sc = new SparkContext(conf)
    val lines = sc.textFile("E:/Spark/天池大数据/data_format1/100W.txt")
    /*
    sortByKey参数有二个。1、true（升序），否则反之。2、标识分片数（分区数）
    flatMap相当于获取一个朋友对象的列表。
     */
    val worldcounts=lines.flatMap(line => line.split(" ")).map(word => (word,1)).reduceByKey((a,b) => a+b).map{case (key,value)=>(value,key)}.sortByKey(true,1)
    worldcounts.foreach(println)
  }
}</span>
           

        Map()和flatMap()区别：

<span style="font-size:18px;"><span style="font-size:18px;">object yb{
   def main (args: Array[String]) {
     val m=List(List(1,2),List(3,4))
     println(m.map(x=>x))
     println(m)
     val x=m.flatten
     println(x)
     println(m.flatMap(x =>x))
  }</span></span>
           

      由上面程序可得flatMap是由Map和flatten综合而来，同时也可以发现flatMap最终都会输出一串的序列，而Map输出是多个集合。

       union()用法：

<span style="font-size:18px;"><span style="font-size:18px;">object yb{
  def main(args: Array[String]) {
    val m1=List(List(1,2),List(3,4))
    val m2=List(List(1,2),List(3,4))
    val unionx=m1.union(m2)//把两个数据集联合起来
    println(unionx)
    val mx1=List(1,2)
    val mx2=List(3,4)
    val unionxx=mx1.union(mx2)//把两个数据集联合起来
    println(unionxx)
  }
}</span></span>
           

      笛卡尔积cartesian()用法：

<span style="font-size:18px;"><span style="font-size:18px;">object yb{
  def main(args: Array[String]) {
    val conf=new SparkConf().setAppName("Test").setMaster("local[4]")
    val sc = new SparkContext(conf)
    val data1=sc.parallelize(List(1,2,3))//并行化，因为笛卡尔积是操作在RDD上的，所以必须是RDD的数据。
    val data2=sc.parallelize(List(4,5,6))
    data1.cartesian(data2).foreach(println)
  }
}</span></span>
           

      groupByKey()和 reduceByKey()区别：

<span style="font-size:18px;"><span style="font-size:18px;">object yb {
  def main(args: Array[String]){
    val conf=new SparkConf().setAppName("Test").setMaster("local[4]")
    val sc = new SparkContext(conf)
    val lines = sc.textFile("E:/Spark/天池大数据/data_format1/100W.txt")
    /*
    sortByKey参数有二个。1、true（升序），否则反之。2、标识分片数（分区数）
     */
    val worldcounts=lines.flatMap(line => line.split(" ")).map(word => (word,1)).reduceByKey((a,b) => a+b).map{case (key,value)=>(value,key)}.sortByKey(false,1)//按照从大到小的顺序排序
    val topK=worldcounts.top(10)
    topK.foreach(println)//输出排名前十的词频
   
  }
}</span></span>
           

    groupByKey不在本地merge统一在主节点merge

    reduceByKey在本地merge然后在到主节点merge

      reduce ()用法：

<span style="font-size:18px;"><span style="font-size:18px;">object yb{
  def main(args: Array[String]) {
    val data=List(1,2,3,4)
   val sum=data.reduce((x,y)=>x+y)
    println(sum)//输出：10
  }
}//reduce将RDD中元素两两传递给输入函数，同时产生一个新的值，新产生的值与RDD中下一个元素再被传递给输入函数直到最后只有一个值为止
    其本质相当于一个满二叉树的左右孩子相加赋给根节点的过程。</span></span>