import org.apache.spark.rdd.RDD

    import org.apache.spark.{SparkConf, SparkContext}

    

    object CreateRDD {

  def main(args: Array[String]): Unit = {

    //每次上传到服务器调试太麻烦了

    //本地调试

    val conf = new SparkConf()

      .setAppName("ScalaWordCount")

      //local 本地运行 - 一个结果文件

      //local[*] 按照cpu的核数开启对应的线程去执行（可以为*个Executor在运行），多个结果文件

      //local[2] 开启2个线程去跑（可以为2个Executor在运行），2个结果文件

      .setMaster("local[2]")

    

    //创建Spark执行入口 - RDD

    val sc = new SparkContext(conf)

    

    

    /*

      创建RDD：

         编程创建方式：parallelize()和makeRDD()创建

         RDD来源创建方式：

            Scala集合中创建：sc.parallelize(List(("jetty",2),("kitty",2),("tom",2)))

            读取外部文件：sc.textFile("hdfs://hadoop100/wc")

            从另一个RDD中创建：val rdd2: RDD[String] = rdd1.flatMap(_.split(" "))

     */

    //从Scala集合中创建

    val rdd:RDD[(String,Int)] = sc.parallelize(List(("jetty",2),("kitty",2),("tom",2)))

    //读取外部文件

    val rdd1: RDD[String] = sc.textFile("hdfs://hadoop100/wc")

    //从另一个集合中获取

    val rdd2: RDD[String] = rdd1.flatMap(_.split(" "))

    

    /*

      底层：

          def makeRDD[T: ClassTag](

            seq: Seq[T],

            numSlices: Int = defaultParallelism): RDD[T] = withScope {

            parallelize(seq, numSlices)

          }

     */

    sc.makeRDD(List(("jetty",2),("kitty",2),("tom",2)))

  }

}

转换函数

描述

map(func)

返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成

filter(func)

返回一个新的RDD，该RDD经过func函数计算后返回值为true的输入元素组成

flatMap(func)

类似于map，但是每个输入元素可以被映射为0或多个输出元素

mapPartitions

类似map，但是独立在RDD的每一个分区运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]，假设有N个元素，有M个分区，那么map的函数的将被调用N次，而mapPartitions被调用M次，一次处理一个分区的所

有数据。

mapPartitionsWithIndex(func)

类似mapPartitions，但func带有一个整数参数，表示分片的索引值，因此在类型为T的RDD上运行时，func函数类型必须为((Int, Interator[T]) => Iterator[U])

taskSample

返回最终的结果集合

union(otherDataset)

求并集

intersection(otherDataset)

求交集

distinct([numTasks])

去重，默认只有8个并行任务，但可以传入一个可选的numTask参数改变

partitionBy

对RDD进行分区操作，如果原有的partitionRDD和现有的一致，就不分区，否则产生ShuffleRDD

reduceByKey(func,[numTasks])

在一个（K,V）的RDD上调用，把Key值聚合到一起，reduce任务通过可选的第二个参数进行设置

groupByKey

对每个key进行操作，但只生成一个sequence

combineByKey(

createCombiner:V => C,

mergeValue(C,V) => C,

mergeCombiners(C,C)=>C)

对相同K，把V合并成为一个集合

createCombiner：会遍历分区所有元素

mergeValue：如果是一个在处理当前分区之前已经遇到的键，会使用mergeValue()方法，将该键的累加器对应的当前值与这个新的值进行合并。

mergeCombiners：将各个分区的结果进行合并

aggregateByKey(zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U)

按Key将Value分组合并，合并时将每个value初始值作为第一个func的参数进行计算，返回一个新的KV对，然后第二个func在进行全局聚合

foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

aggregateByKey的简化操作

sortByKey([ascending], [numTasks])

在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD

sortBy(func,[ascending], [numTasks])

与sortByKey类似，但是更灵活,可以用func先对数据进行处理，按照处理后的数据比较结果排序。

join(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

cogroup(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

cartesian(otherDataset)

笛卡尔积

pipe(command, [envVars])

对于每个分区，都执行一个perl或者shell脚本，返回输出的RDD

coalesce(numPartitions)     

缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。

repartition(numPartitions)

根据分区数，从新通过网络随机洗牌所有数据。

repartitionAndSortWithinPartitions(partitioner)

repartitionAndSortWithinPartitions函数是repartition函数的变种，与repartition函数不同的是，repartitionAndSortWithinPartitions在给定的partitioner内部进行排序，性能比repartition要高。 

glom

将每一个分区形成一个数组，形成新的RDD类型时RDD[Array[T]]

mapValues

针对于(K,V)形式的类型只对V进行操作 

subtract

计算差的一种函数去除两个RDD中相同的元素，不同的RDD将保留下来 

转换函数

描述

reduce(func)

通过func函数聚集RDD中的所有元素，这个功能必须是可交换且可并联的

collect()

在驱动程序中，以数组的形式返回数据集的所有元素

count()

返回RDD的元素个数

first()

返回RDD的第一个元素（类似于take(1)）

take(n)

返回一个由数据集的前n个元素组成的数组

takeSample(withReplacement,num, [seed])

返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子

takeOrdered(n)

返回前几个的排序

aggregate (zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)

aggregate函数将每个分区里面的元素通过seqOp和初始值进行聚合，然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作。这个函数最终返回的类型不需要和RDD中元素类型一致。

fold(num)(func)

折叠操作，aggregate的简化操作，seqop和combop一样。

saveAsTextFile(path)

将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本

saveAsSequenceFile(path) 

将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。

saveAsObjectFile(path) 

用于将RDD中的元素序列化成对象，存储到文件中。

countByKey()

针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。

foreach(func)

在数据集的每一个元素上，运行函数func进行更新。

import org.apache.spark.rdd.RDD

    import org.apache.spark.{Partitioner, SparkConf, SparkContext}

    

    object MyPartition {

  def main(args: Array[String]): Unit = {

    //每次上传到服务器调试太麻烦了

    //本地调试

    val conf = new SparkConf()

      .setAppName("ScalaWordCount")

      //local 本地运行 - 一个结果文件

      //local[*] 按照cpu的核数开启对应的线程去执行（可以为*个Executor在运行），多个结果文件

      //local[2] 开启2个线程去跑（可以为2个Executor在运行），2个结果文件

      .setMaster("local[2]")

    

    //创建Spark执行入口 - RDD

    val sc = new SparkContext(conf)

    

    val rdd:RDD[(String,Int)] = sc.parallelize(List(("jetty",2),("kitty",2),("tom",2)))

    

    val rdd2:RDD[(String,Int)] = rdd.reduceByKey(_+_)

    

    rdd2.partitionBy(new MyPartitioner(3,rdd))

    

    rdd2.cache()

    

    println("==================================")

    println(rdd2.collect().toBuffer)

    println(rdd2.partitions.length)

    //释放资源

    sc.stop()

  }

}

    

    //自定义分区器，默认是HashPartitioner

    class MyPartitioner(num: Int,rDD: RDD[(String,Int)]) extends Partitioner {

    

  //分区的数量，下一个RDD有多少个分区

  override def numPartitions: Int = num

    

  //根据传入的key计算分区标号

  //用key去标识分区，可传数组等(Any)

  override def getPartition(key: Any): Int = {

    val sub = key.asInstanceOf[(String,Int)]

    //分区编号

    sub._2

  }

}

import org.apache.hadoop.io.{LongWritable, Text}

    import org.apache.spark.{SparkConf, SparkContext}

    import org.apache.spark.rdd.RDD

    

    object SaveFile {

  def main(args: Array[String]): Unit = {

    //每次上传到服务器调试太麻烦了

    //本地调试

    val conf = new SparkConf()

      .setAppName("ScalaWordCount")

      //local 本地运行 - 一个结果文件

      //local[*] 按照cpu的核数开启对应的线程去执行（可以为*个Executor在运行），多个结果文件

      //local[2] 开启2个线程去跑（可以为2个Executor在运行），2个结果文件

      .setMaster("local[2]")

    

    //创建Spark执行入口 - RDD

    val sc = new SparkContext(conf)

    

    

    //从Scala集合中创建

    val rdd:RDD[(String,Int)] = sc.parallelize(List(("jetty",2),("kitty",2),("tom",2)))

    //读取外部文件

    val rdd1: RDD[String] = sc.textFile("hdfs://hadoop100/wc")

    sc.

    /*

      底层：

          def makeRDD[T: ClassTag](

            seq: Seq[T],

            numSlices: Int = defaultParallelism): RDD[T] = withScope {

            parallelize(seq, numSlices)

          }

     */



    val hdfs: String = "hdfs://hadoop100/save"

    

    //保存对象文件，需要实现序列化接口

    rdd.saveAsObjectFile(hdfs + "Obj")

    rdd.saveAsTextFile(hdfs + ".txt")

    rdd.saveAsTextFile(hdfs + ".json")

    rdd.saveAsHadoopFile(hdfs + "Hadoop")

    rdd.saveAsSequenceFile(hdfs + "Seq")

    

    //这是Hadoop新Api

    //rdd.saveAsNewAPIHadoopFile(hdfs + "NewHadoopApi",classOf[LongWritable],classOf[Text],classOf[org.apache.hadoop.mapreduce.lib.output.TextOutputFormat[LongWritable,Text]])

    //其他的 rdd.saveAs....

    

    //读取

    val rddSeqFile: RDD[(Nothing, Nothing)] = sc.sequenceFile(hdfs + "Seq")

    

  }

}

import java.sql.DriverManager

    

    import org.apache.spark.rdd.{JdbcRDD, RDD}

    import org.apache.spark.{SparkConf, SparkContext}

    

    /**

  * JDBC 连接数据库

  */

    object JdbcRDDTest {

    

  val getConn = () => {

    DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/bigdata?characterEncoding=utf-8","root","root")

  }

    

  def main(args: Array[String]): Unit = {

    

    //配置

    val conf = new SparkConf()

                      .setAppName("JdbcRDD")

                      .setMaster("local[*]")

    

    //程序入口

    val sc = new SparkContext(conf)

    

    //里面没有真正要计算的数据，而是告诉RDD，以后触发Action，在哪里读取

    val jdbcRDD:RDD[(Int,String)] = new JdbcRDD(sc,

                getConn,

                "select * from access_log where id = ? and id = ?",

                1, //第一个?

                3, //upperBound

                2, //分区数量

                rs => { //结果

                  val id = rs.getInt(1)

                  val name = rs.getString(2)

                  (id,name)

                })

    

    //触发action

    val r = jdbcRDD.collect()

    println(r.toBuffer)

    

    //释放资源

    sc.stop()

  }

}

import java.sql.{Connection, DriverManager, PreparedStatement}

    

    import org.apache.spark.broadcast.Broadcast

    import org.apache.spark.rdd.RDD

    import org.apache.spark.{SparkConf, SparkContext}

    

    

    

/**

  * 广播变量：是用来高效的分发一些较大的对象，

  * 向所有工作节点发送一个只读值，以供一个或多个Spark操作使用。

  * 比如：需要向所有节点发送一个较大的只读查询表，

  * 甚至是机器学习算法中的一个很大特征向量，广播变量用起来会很顺手。

  */

    object Broadcase {

  def main(args: Array[String]): Unit = {

    

    //Spark配置

    val conf = new SparkConf()

                      .setAppName("Broadcase")

                      .setMaster("local[*]")

    

    //Spark应用入口

    val sc = new SparkContext(conf)

    

    val rdd:RDD[(String,Int)] = sc.parallelize(List(("jetty",2),("kitty",2),("tom",2)))

    

    //一个大的变量，不用在每一个都有一份，直接公用一个大的变量

    val broadcase: Broadcast[Array[(Int,String)]] = sc.broadcast(Array((1,"jerry"),(2,"tom"),(3,"mike")))

    

    

    //对于性能的优化，可以使用这个，每次提取的是一整个分区数据做批量处理，适合多数据使用

    //视场景使用

    //rdd.foreachPartition()

    

    //-----------------------------------------

    // 没有实现Serializable接口，传送到Executor执行时就执行不了。

    //这也说明了，到Executor执行的对象，必须实现Seriablizable接口

    

    //写入到数据库中

    //val connection: Connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/bigdata?characterEncoding=utf-8","root","root")

    //保存数据

    //val prepareStatement: PreparedStatement = connection.prepareStatement("INSERT INTO access_log VALUES(?,?)")

    //-----------------------------------------

    

    

    rdd.foreach(tp2 => {

      //只有是jetty的话，才需要把数据保存到数据库

      if(tp2._1.equals("jetty")){

        //取出广播变量值

        //广播变量，广播出去的内容就不能改变了，如果需要实时改变可以把规则放到Redis中

        val value: Array[(Int,String)] = broadcase.value

    

        //写入到数据库中

        val connection: Connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/bigdata?characterEncoding=utf-8","root","root")

    

        //保存数据

        val prepareStatement: PreparedStatement = connection.prepareStatement("INSERT INTO access_log VALUES(?,?)")

    

        value.foreach(tp2 => {

          prepareStatement.setInt(1,tp2._1) //第一个 ?

          prepareStatement.setString(2,tp2._2) //第二个 ?

          prepareStatement.executeUpdate()

        })

    

        //释放资源

        prepareStatement.close()

        connection.close()

      }

    })

    

    //释放资源

    sc.stop()

  }

}

import org.apache.spark.rdd.RDD

    import org.apache.spark.{SparkConf, SparkContext}

    

    object MySort {

  def main(args: Array[String]): Unit = {

    

    //配置

    val conf = new SparkConf()

                      .setAppName("MySort")

                      .setMaster("local[*]")

    

    //程序入口

    val sc = new SparkContext(conf)

    

    //排序规则：首先按照颜值从高到低进行排序，如果颜值相等，在按照年龄升序

    val users = Array("A 30 92","B 28 98","C 32 99")

    

    //将Driver端的数据变成RDD

    val lines:RDD[String] = sc.parallelize(users)

    

    //切分整理数据

    /*val tpRDD:RDD[(String,Int,Int)] = lines.map(line => {

      val fields = line.split(" ")

      val name = fields(0)

      val age = fields(1).toInt

      val fv = fields(2).toInt

      (name,age,fv)

    })*/

    

    //不满足要求

    //tpRDD.sortBy(tpRDD => tpRDD._3,false)

    

    val tpRDD:RDD[User] = lines.map(line => {

      val fields = line.split(" ")

      val name = fields(0)

      val age = fields(1).toInt

      val fv = fields(2).toInt

      new User(name,age,fv)

    })

    

    //将RDD内的User类型数据进行排序

    val sorted: RDD[User] = tpRDD.sortBy(u => u)

    

    val r = sorted.collect()

    println("=========================")

    println(r.toBuffer)

    

    //释放资源

    sc.stop()

  }

}

    

    class User(val name: String,val age: Int,val fv: Int) extends Ordered[User] with Serializable {

    

  override def compare(that: User): Int = {

    //正数是升序，负数，你懂得

    if(this.fv == that.fv) {

      this.age - that.age

    } else {

      -(this.fv - that.fv)

    }

  }

    

  override def toString: String = {

    name + " - " + age + " - " + fv

  }

}

/**
 * Smarter version of hadoopFile() that uses class tags to figure out the classes of keys,
 * values and the InputFormat so that users don't need to pass them directly. Instead, callers
 * can just write, for example,
 * {{{
 * val file = sparkContext.hadoopFile[LongWritable, Text, TextInputFormat](path)
 * }}}
 *
 * @note Because Hadoop's RecordReader class re-uses the same Writable object for each
 * record, directly caching the returned RDD or directly passing it to an aggregation or shuffle
 * operation will create many references to the same object.
 * If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first
 * copy them using a `map` function.
 * @param path directory to the input data files, the path can be comma separated paths as
 * a list of inputs
 * @return RDD of tuples of key and corresponding value
 */
def hadoopFile[K, V, F <: InputFormat[K, V]](path: String)
    (implicit km: ClassTag[K], vm: ClassTag[V], fm: ClassTag[F]): RDD[(K, V)] = withScope {
  hadoopFile[K, V, F](path, defaultMinPartitions)
}

/**
 * Smarter version of `newApiHadoopFile` that uses class tags to figure out the classes of keys,
 * values and the `org.apache.hadoop.mapreduce.InputFormat` (new MapReduce API) so that user
 * don't need to pass them directly. Instead, callers can just write, for example:
 * ```
 * val file = sparkContext.hadoopFile[LongWritable, Text, TextInputFormat](path)
 * ```
 *
 * @note Because Hadoop's RecordReader class re-uses the same Writable object for each
 * record, directly caching the returned RDD or directly passing it to an aggregation or shuffle
 * operation will create many references to the same object.
 * If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first
 * copy them using a `map` function.
 * @param path directory to the input data files, the path can be comma separated paths
 * as a list of inputs
 * @return RDD of tuples of key and corresponding value
 */
def newAPIHadoopFile[K, V, F <: NewInputFormat[K, V]]
    (path: String)
    (implicit km: ClassTag[K], vm: ClassTag[V], fm: ClassTag[F]): RDD[(K, V)] = withScope {
  newAPIHadoopFile(
    path,
    fm.runtimeClass.asInstanceOf[Class[F]],
    km.runtimeClass.asInstanceOf[Class[K]],
    vm.runtimeClass.asInstanceOf[Class[V]])
}