Spark/Scala note

Spark 常用代码段和Scala的一些笔记

Spark 常用代码段

创建SparkContext

val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
val sc = new SparkContext(conf)
sc.setLogLevel("WARN")

sc.stop()

创建SparkSession

val spark = SparkSession
.builder()
.appName("Demo1")
.master("local[*]")
.enableHiveSupport()
// Spark使用与Hive相同的约定写parquet数据
.config("spark.sql.parquet.writeLegacyFormat", "true")
.getOrCreate()
val sc = spark.sparkContext
sc.setLogLevel("warn")

// For implicit conversions like converting RDDs to DataFrames 
import spark.implicits._

spark.close()

SparkSession从csv读取

val df: DataFrame = spark.read
.option("header", "true")
.option("inferschema", "true")
.csv("data/emp.dat")

df.printSchema()
df.show()

SparkSession执行sql样例

// 准备数据
val arr = Array("1 1,2,3", "2 2,3", "3 1,2")
val rdd: RDD[Info] = spark.sparkContext.makeRDD(arr)
.map { line =>
  val fields: Array[String] = line.split("\\s+")
  Info(fields(0), fields(1))
}
val ds: Dataset[Info] = spark.createDataset(rdd)
ds.createOrReplaceTempView("t1")
ds.show

// 用SQL处理 - HQL
spark.sql(
  """
        |select id, tag
        |  from t1
        |       lateral view explode(split(tags, ",")) t2 as tag
        |""".stripMargin
).show

// SparkSQL
spark.sql(
  """
        |select id, explode(split(tags, ",")) tag
        |  from t1
        |""".stripMargin
).show

SparkSession写入csv

spark.sql("select * from people") 
.write .format("csv") 
.mode("overwrite") 
.save("data/csv")

SparkSession连接jdbc

// jdbc
val jdbcDF: DataFrame = spark.read
.format("jdbc")
.option("url", "jdbc:mysql://linux123:3306/ebiz?useSSL=false")
.option("user", "hive")
.option("password", "12345678")
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtable", "lagou_product_info")
.load()
jdbcDF.show()

jdbcDF.write
.format("jdbc")
.option("url", "jdbc:mysql://linux123:3306/ebiz?useSSL=false&characterEncoding=utf8")
.option("user", "hive")
.option("password", "12345678")
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtable", "lagou_product_info_back")
.mode(SaveMode.Append)
.save()

创建StreamingContext

// 初始化
Logger.getLogger("org").setLevel(Level.ERROR)
val conf: SparkConf = new SparkConf()
.setAppName(this.getClass.getCanonicalName)
.setMaster("local[2]")
val ssc = new StreamingContext(conf, Seconds(10))

// 创建DStream
val lines: DStream[String] = ssc.socketTextStream("localhost", 9999)

// DStream转换
val words: DStream[String] = lines.flatMap(_.split("\\s+"))
val result: DStream[(String, Int)] = words.map((_, 1)).reduceByKey(_ + _)

// DStream输出
result.print(20)

// 启动作业
ssc.start()
ssc.awaitTermination()

StreamingContext整合Kafka

object KafkaDStream1 {
  def main(args: Array[String]): Unit = {
    // 初始化
//    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf = new SparkConf().setAppName("FileDStream").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    // 定义kafka相关参数
    val groupId: String = "mygroup01"
    val topics: Array[String] = Array("lg_bus_info2")
    val kafkaParams: Map[String, Object] = getKafkaConsumerParameters(groupId)

    // 创建DStream
    val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )

    // DStream转换&输出
    dstream.foreachRDD{ (rdd, time) =>
      println(s"*********** rdd.count = ${rdd.count()}; time = $time *************")
    }

    // 启动作业
    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaConsumerParameters(groupid: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop3:9092,hadoop4:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupid,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean),
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
    )
  }
}

Spark GraphX基本操作

object GraphXExample1 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")

    // 定义顶点
    val vertexArray: Array[(VertexId, (String, Int))] = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50))
    )
    val vertexRDD: RDD[(VertexId, (String, Int))] = sc.makeRDD(vertexArray)

    // 定义边
    val edgeArray: Array[Edge[Int]] = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 6),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3)
    )
    val edgeRDD: RDD[Edge[Int]] = sc.makeRDD(edgeArray)

    // 图的定义
    val graph: Graph[(String, Int), Int] = Graph(vertexRDD, edgeRDD)

    // 属性操作(找出图中年龄 > 30 的顶点；属性 > 5 的边； 属性 > 5 的 triplets)
//    graph.vertices
//        .filter{case (_, (_, age)) => age > 30}
//        .foreach(println)
//
//    graph.edges
//        .filter(edge => edge.attr > 5)
//      .foreach(println)
//
//    graph.triplets
//      .filter { t => t.attr > 5 }
//      .foreach(println)

    // 属性操作。degress操作，找出图中最大的出度、入度、度数
//    val inDegress: (VertexId, Int) = graph.inDegrees
//      .reduce((x, y) => if (x._2 > y._2) x else y)
//    println(s"inDegress = $inDegress")
//
//    val outDegress: (VertexId, Int) = graph.outDegrees
//      .reduce((x, y) => if (x._2 > y._2) x else y)
//    println(s"outDegress = $outDegress")
//
//    val degress: (VertexId, Int) = graph.degrees
//      .reduce((x, y) => if (x._2 > y._2) x else y)
//    println(s"degress = $degress")

    // 转换操作。顶点转换，所有人年龄加 100
//    graph.mapVertices{case (id, (name, age)) => (id, (name, age+100))}
//        .vertices
//        .foreach(println)

    // 边的转换，边的属性*2
//    graph.mapEdges(e => e.attr * 2)
//        .edges
//        .foreach(println)

    // 结构操作。顶点年龄 > 30 的子图
//    val subGraph: Graph[(String, Int), Int] = graph.subgraph(vpred = (id, vd) => vd._2 > 30)
//    subGraph.edges.foreach(println)
//    subGraph.vertices.foreach(println)

    // 找出出度=入度的人员。连接操作
    // 思路：图 + 顶点的出度 + 顶点的入度 => 连接操作
//    val initailUserGraph: Graph[User, Int] = graph.mapVertices { case (id, (name, age)) => User(name, age, 0, 0) }
//
//    val userGraph: Graph[User, Int] = initailUserGraph.outerJoinVertices(initailUserGraph.inDegrees) {
//      case (id, u, inDeg) => User(u.name, u.age, inDeg.getOrElse(0), u.outDegress)
//    }.outerJoinVertices(initailUserGraph.outDegrees) {
//      case (id, u, outDeg) => User(u.name, u.age, u.inDegress, outDeg.getOrElse(0))
//    }
//    userGraph.vertices.filter{case (_, user) => user.inDegress==user.outDegress}
//        .foreach(println)


    // 顶点5到其他各顶点的最短距离。聚合操作(Pregel API)
    val sourceId: VertexId = 5L
    val initailGraph: Graph[Double, Int] = graph.mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity)

    val disGraph: Graph[Double, Int] = initailGraph.pregel(Double.PositiveInfinity)(
      // 两个消息来的时候，取其中的最小路径
      (id, dist, newDist) => math.min(dist, newDist),

      // Send Message 函数
      triplet => {
        if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
          Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
        } else
          Iterator.empty
      },

      // mergeMsg
      (dista, distb) => math.min(dista, distb)
    )

    disGraph.vertices.foreach(println)

    sc.stop()
  }
}

case class User(name: String, age: Int, inDegress: Int, outDegress: Int)

Spark GraphX从文件生成图，生成连通图

object GraphXExample2 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")

    // 生成图
    val graph: Graph[Int, Int] = GraphLoader.edgeListFile(sc, "data/graph.dat")

    graph.vertices.foreach(println)
    graph.edges.foreach(println)

    // 调用连通图算法
    graph.connectedComponents()
        .vertices
      .sortBy(_._2)
        .foreach(println)

    sc.stop()
  }
}

Scala基础

List

简介

List代表元素顺序固定的不可变的链表，它是Seq的子类，在Scala编程中经常使用。

List列表有头部和尾部的概念，可以分别使用head和tail方法来获取：

• head返回的是列表第一个元素的值
• tail返回的是除第一个元素外的其它元素构成的新列表

Scala定义了一个空列表对象Nil，定义为List[Nothing]

借助 Nil 可将多个元素用操作符 :: 添加到列表头部，常用来初始化列表；

操作符 ::: 用于拼接两个列表

// 不可变集合
    val list1: List[Int] = 1::2::Nil
    println(list1)
    println(list1.head)
    println(list1.tail)
//    List(1, 2)
//    1
//    List(2)
    val list2: List[Int] = 3::4::Nil
    val list3: List[Int] = list1:::list2
    println(list3)
//    List(1, 2, 3, 4)

快速排序

def quickSort(lst:List[Int]):List[Int]= {
    lst match {
      case Nil=>Nil
      case head::tail=>{
        val (less,greater) = tail.partition((_: Int) < head)
        quickSort(less):::(head::quickSort(greater))
      }
    }
  }

可变list

import scala.collection.mutable.ListBuffer

获取元素（使用括号访问(索引值)）
添加元素（+=）
追加一个列表（++=）
更改元素（使用括号获取元素，然后进行赋值）
删除元素（-=）
转换为List（toList）
转换为Array（toArray)

    // 可变集合
    import scala.collection.mutable.ListBuffer
    val buffer: ListBuffer[Int] = ListBuffer()
    buffer+=3
    buffer+=10
    buffer+=1
    buffer+=6
    println(buffer)
    buffer++=ListBuffer(2,3,15,4)
    println(buffer)
    println(buffer(0))
//    ListBuffer(3, 10, 1, 6)
//    ListBuffer(3, 10, 1, 6, 2, 3, 15, 4)
//    3s

Scala Iterator（迭代器）

简介

• Scala Iterator（迭代器）不是一个集合，它是一种用于访问集合的方法。
• 迭代器 it 的两个基本操作是 next 和 hasNext。
• 调用 it.next() 会返回迭代器的下一个元素，并且更新迭代器的状态。
• 调用 it.hasNext() 用于检测集合中是否还有元素。

val lst = List(1,2,3,4)
  val iterator: Iterator[Int] = lst.iterator
  while(iterator.hasNext) {
    println(iterator.next())
  }

Option的类型

Option有两个子类别，Some和None。当程序回传Some的时候，代表这个函式成功地给了你一个String，而你可以透过get()函数拿到那个String，如果程序返回的是None，则代表没有字符串可以给你。

object Option_Test {
  def main(args: Array[String]): Unit = {
    val grade = Map("a"->90,"b"->60,"c"->80)
    val temp: Int = grade("a")
    println(temp)
    // println(grade("q")) error
    def getGrade(name:String)={
      val maybeInt: Option[Int] = grade.get(name)
      maybeInt match {
        case None=>println("None")
        case Some(maybeInt)=>println(maybeInt)
      }
    }
    getGrade("b")
    getGrade("d")
  }
}

Map类型

如果要使用可变Map，必须导入scala.collection.mutable.Map

    val stringToInt = Map("a" -> 50, "b" -> 60, "c" -> 80)
    val stringToInt2 = Map(("a",50), ("b",60), ("c",80))
    println(stringToInt)
    println(stringToInt2)
    println(stringToInt.get("a"))
    println(stringToInt.get("q"))
    println(stringToInt.getOrElse("q",0))
//    Map(a -> 50, b -> 60, c -> 80)
//    Map(a -> 50, b -> 60, c -> 80)
//    Some(50)
//    None
//    0

可变map

    // 可变map
    val b: mutable.Map[String, Int] = scala.collection.mutable.Map("a" -> 1, "b" -> 2, "c" -> 3)
    b("a") = 2
    // 增加了一个新元素
    b("d") = 4
    // 用 + 添加新的元素；用 – 删除元素
    b += ("e" -> 1, "f" -> 2)
    b -= "a"
    println(b)  // 乱序
//    Map(e -> 1, b -> 2, d -> 4, c -> 3, f -> 2)

拉链操作创建map

    // 拉链操作创建map
    val p: Array[Int] = Array(1,2,3)
    val q: Array[String] = Array("a","b","c")
    val tuples: Array[(Int, String)] = p.zip(q)
    val map: scala.collection.immutable.Map[Int, String] = p.zip(q).toMap
    println(tuples)
    println(map)
//    Map(e -> 1, b -> 2, d -> 4, c -> 3, f -> 2)
//    [Lscala.Tuple2;@51e2adc7 
//    Map(1 -> a, 2 -> b, 3 -> c)

Array

开始学的时候有些不重视，花招还是挺多的

数组可索引、类型一致、长度不变

    val nums: Array[Int] = new Array[Int](10)
    nums(8) = 10
    println(nums.toBuffer)

    val nums2: Array[Int] = Array(1,2,3,5,6)
    println(nums2.mkString(","))

    val nums3: Array[Int] = (1 to 10).toArray
    println(nums3.toBuffer)
//    ArrayBuffer(0, 0, 0, 0, 0, 0, 0, 0, 10, 0)
//    1,2,3,5,6
//    ArrayBuffer(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

变长数组ArrayBuffffer

    // 变长数组
    val num: ArrayBuffer[Int] = ArrayBuffer[Int]()
    num+=1
    num++=ArrayBuffer(1,2,3,4)
    println(num)
    num.append(100,101)
    println(num)
    num.insert(0,5,5)
    println(num)
    // 移除最后2元素
    num.trimEnd(2)
    // 移除最开始的一个或者多个元素
    num.trimStart(1)
    // 从下标2处移除一个或者多个元素
    println(num)
    num.remove(2,2)
    println(num)
//    ArrayBuffer(1, 1, 2, 3, 4)
//    ArrayBuffer(1, 1, 2, 3, 4, 100, 101)
//    ArrayBuffer(5, 5, 1, 1, 2, 3, 4, 100, 101)
//    ArrayBuffer(5, 1, 1, 2, 3, 4)
//    ArrayBuffer(5, 1, 3, 4)

多维数组Array.ofDim

通过Array的ofDim方法来定义一个多维的数组，多少行，多少列，都是自己说了算

    // 3*4数组
    val dim: Array[Array[Int]] = Array.ofDim[Int](3,4)
    dim(1)(1) = 100;
    for(i<-0 to 2;j<-0 to 3) {
      print(dim(i)(j)+" ")
      if(j==3) println()
    }
//    0 0 0 0
//    0 100 0 0
//    0 0 0 0

元组tuple

元组是不同类型的值的集合，元组中的元素可以是不同的数据类型，元组的元素个数上限是22个

Tuple的访问形式比较特殊。元组的下标从1开始

object Tuple_Test {
  def main(args: Array[String]): Unit = {
    val a: (Int, Double, String, Char) = (1, 1.2, "ad", 'd')
    val b: (Int, Double, String, Char) = Tuple4(1, 1.2, "ad", 'd')
    println(a==b)   // true
    println(a._1)   // 1
    a.productIterator.foreach(println)  // loop
  }
}

None,Nothing,Null,Nil

Null是所有AnyRef的子类，在scala的类型系统中，AnyRef是Any的子类，同时Any子类的还有AnyVal。对应java值类型的所有类型都是AnyVal的子类。所以Null可以赋值给所有的引用类型(AnyRef)，不能赋值给值类型，这个java的语义是相同的。 null是Null的唯一对象

Nothing是所有类型的子类，它没有对象，但是可以定义类型，如果一个类型抛出异常，那这个返回值类型就是Nothing

Nil是一个空的List，定义为List[Nothing]，根据List的定义List[+A]，所有Nil是所有List[T]的子类。

None是一个object，是Option的子类型

类与对象

了解一下

简介

在Scala中，类并不用声明为public；

Scala源文件中可以包含多个类，所有这些类都具有公有可见性；

val修饰的变量（常量），值不能改变，只提供getter方法，没有setter方法；

var修饰的变量，值可以改变，对外提供getter、setter方法；

如果没有定义构造器，类会有一个默认的无参构造器；

基础类的注意事项

Scala中声明一个字段，必须显示的初始化，然后根据初始化的数据类型自动推断其类型，字段类型可以省略

_ 表示一个占位符，编译器会根据变量的数据类型赋予相应的初始值

使用占位符，变量类型必须指定

_ 对应的默认值:整型默认值0；浮点型默认值0.0；String与引用类型，默认值null; Boolean默认值false

val修饰的变量不能使用占位符

class Person {
  var name = "xiaoyuyu"
  var nickName:String = _
  var age =20
  // 类私有字段,有私有的getter方法和setter方法
  // 在类的内部可以访问，其伴生对象也可以访问
  private var hobby:String = _
  // 对象私有字段,访问权限更加严格，只能在当前类中访问
  private[this] val pwd = "123456"

  def hello(): Unit = {
    println("hello " + name)
  }
}

object class_Test {
  def main(args: Array[String]): Unit = {
    val person = new Person()
    //注意：如果使用对象的属性加上 _= 给var修饰的属性进行重新赋值，其实就是调用age_=这个setter方法
    person.name_=("jack")
    person.age = 23
    //直接调用类的属性，其实就是调用getter方法
    println(person.age)
    person.hello()
  }
}

自定义getter setter

class Dog { 
	private var _leg = 0 

  //自定义getter方法 
  def leg = _leg 

	//自定义setter方法 
  def leg_=(newLeg: Int) { 
		_leg = newLeg 
	}
}

// 使用自定义getter和setter方法 
val dog = new Dog 
dog.leg_=(4) 
println(dog.leg)

bean属性

JavaBean规范把Java属性定义为一堆getter和setter方法。

类似于Java，当将Scala字段标注为 @BeanProperty时，getFoo和setFoo方法会自动生成。

使用@BeanProperty并不会影响Scala自己自动生成的getter和setter方法。

在使用时需要导入包scala.beans.BeanProperty

构造器

如果没有定义构造器，Scala类中会有一个默认的无参构造器；Scala当中类的构造器分为两种：主构造器和辅助构造器；

主构造器的定义与类的定义交织在一起，将主构造器的参数直接放在类名之后。

当主构造器的参数不用var或val修饰时，参数会生成类的私有val成员。

Scala中，所有的辅助构造器都必须调用另外一个构造器，另外一个构造器可以是辅助构造器，也可以是主构造器。

//主构造器直接定义在类中，其代码不包含在任何方法中
// Scala中的主构造器与类名交织在一起，类名后面的参数即为主构造器的参数
class Dog(name:String,var age:Int){
  //类中不在任何方法中的代码，都属于主构造器的代码。
  // 创建类的对象时会去执行主构造器的代码。下面的println代码就是主构造器的一部分
  println(name)
  println(age)

  var gender:String = _

  def this(name:String,age:Int,gender: String) {
    //每个辅助构造器，都必须以其他辅助构造器，或者主构造器的调用作为第一句代码
    this(name,age)
    this.gender = gender
  }

  var color:String = _
  def this(name: String, age: Int, gender: String, color: String) {
    //调用上面的辅助构造器
    this(name, age, gender)
    this.color = color
  }
}

object Dog {
  def main(args: Array[String]): Unit = {
    val dog1=new Dog("狗蛋",4)
    val dog2=new Dog("旺才",3,"雄性")
    val dog3=new Dog("小六",5,"雄性","黑色")
//    狗蛋
//    4
//    旺才
//    3
//    小六
//    5
  }
}

对象

单例对象

java中单例模式主要用到了静态，但是scala中没有static关键字，但是scala可以用伴生类于伴生对象来模拟静态，但首先我们需要了解单例对象

Scala中的单例对象具有如下特点：

• 1、创建单例对象不需要使用new关键字
• 2、object中只有无参构造器
• 3、主构造代码块只能执行一次，因为它是单例的

object ObjectDemo { 
  println("这是单例对象的代码！") 
  
  def main(args: Array[String]): Unit = { 
    val object1=ObjectDemo 
    val object2=ObjectDemo 
  } 
}

伴生类与伴生对象

当单例对象与某个类具有相同的名称时，它被称为这个类的“伴生对象”；

类和它的伴生对象必须存在于同一个文件中，而且可以相互访问私有成员（字段和方法）；

个人觉得就是为了弥补没有static关键字的手段

class static_test{
  val id=1
  private val name="xiaoyuyu"
  def printName(): Unit ={
    println(static_test.action + name)
  }
}

object static_test {
  private val action = "run"
  def main(args: Array[String]): Unit = {
    val p = new static_test
    p.printName() // runxiaoyuyu
  }
}

object中的apply方法

object 中有一个非常重要的特殊方法 – apply方法；

• apply方法通常定义在伴生对象中，目的是通过伴生类的构造函数功能，来实现伴生对象的构造函数功能；
• 通常我们会在类的伴生对象中定义apply方法，当遇到类名(参数1,…参数n)时apply方法会被调用；
• 在创建伴生对象或伴生类的对象时，通常不会使用new class/class() 的方式，而是直接使用 class()隐式的调用伴生对象的 apply 方法，这样会让对象创建的更加简洁；

abstract class Animal { 
  def speak 
}

class Dog extends Animal { 
  override def speak: Unit = { 
    println("woof") 
  } 
}

class Cat extends Animal { 
  override def speak: Unit = { 
    println("meow") 
	} 
}

object Animal { 
  def apply(str: String): Animal = { 
    if (str == "dog") 
    new Dog 
    else
    new Cat 
	}

	def main(args: Array[String]): Unit = { 
  	val cat = Animal("cat") 
  	cat.speak 
  	val dog = Animal("dog") 
  	dog.speak
	} 
}

继承

这个和java差不多，依然是extends

类型检查与转换

要测试某个对象是否属于某个给定的类，可以用isInstanceOf方法。如果测试成功，可以用asInstanceOf方法进行类型转换。

if(p.isInstanceOf[Employee]){ 
	//s的类型转换为Employee 
  val s = p.asInstanceOf[Employee] 
}

如果p指向的是Employee类及其子类的对象，则p.isInstanceOf[Employee]将会成功。

如果p是null，则p.isInstanceOf[Employee]将返回false，且p.asInstanceOf[Employee]将返回null。

如果p不是一个Employee，则p.asInstanceOf[Employee]将抛出异常。

如果想要测试p指向的是一个Employee对象但又不是其子类，可以用：

if(p.getClass == classOf[Employee])

不过，与类型检查和转换相比，模式匹配通常是更好的选择

p match{
  //将s作为Employee处理 
  case s: Employee => ... 
  //p不是Employee的情况 
  case _ => .... 
}

特质trait

作为接口trait

Scala中的trait特质是一种特殊的概念。

首先可以将trait作为接口来使用，此时的trait就与Java中的接口非常类似。

在trait中可以定义抽象方法，与抽象类中的抽象方法一样，只要不给出方法的具体实现即可。

类可以使用extends关键字继承trait。

注意：在Scala中没有implement的概念，无论继承类还是trait特质，统一都是extends。

类继承trait特质后，必须实现其中的抽象方法，实现时可以省略override关键字。

Scala不支持对类进行多继承，但是支持多重继承trait特质，使用with关键字即可。

//定义一个trai特质 

trait HelloTrait { 
  def sayHello 
}

//定义一个trai特质 

trait MakeFriendTrait { 
  def makeFriend 
}

//继承多个trait,第一个trait使用extends关键字，其它trait使用with关键字 
class Person(name: String) extends HelloTrait with MakeFriendsTrait with Serializable { 
  override def sayHello() = println("Hello, My name is " + name) 
  //override关键字也可以省略 
  def makeFriend() = println("Hello，" + name) 
}

带有具体实现的特质

• Scala中的trait特质不仅仅可以定义抽象方法，还可以定义具体实现的方法，这时的trait更像是包含了通用工具方法的类。比如，trait中可以包含一些很多类都通用的功能方法，比如打印日志等等，Spark中就使用了trait来定义通用的日志打印方法。
• Scala trait特质中的字段可以是抽象的，也可以是具体的。

特质构造顺序

在Scala中，trait特质也是有构造器的，也就是trait中的不包含在任何方法中的代码。

构造器以如下顺序执行：

• 1、执行父类的构造器；
• 2、执行trait的构造器，多个trait从左到右依次执行；
• 3、构造trait时会先构造父trait，如果多个trait继承同一个父trait，则父trait只会构造一次；
• 4、所有trait构造完毕之后，子类的构造器才执行

class Person2 { println("Person's constructor!") } 

trait Logger { println("Logger's constructor!") } 

trait MyLogger extends Logger { println("MyLogger's constructor!") } 

trait TimeLogger extends Logger { println("TimeLogger's constructor!") } 

//类既继承了类又继承了特质，要先写父类 
class Student2 extends Person2 with MyLogger with TimeLogger { 
  println("Student's constructor!") 
}

// Person's constructor! 
// Logger's constructor! 
// MyLogger's constructor! 
// TimeLogger's constructor! 
// Student's constructor!

特质继承类

在Scala中，trait特质也可以继承class类，此时这个class类就会成为所有继承此trait的类的父类。

排序Ordered和Ordering

在Java中对象的比较有两个接口，分别是Comparable和Comparator。它们之间的区别在于：

实现Comparable接口的类，重写compareTo()方法后，其对象自身就具有了可比较性； 实现Comparator接口的类，重写了compare()方法后，则提供一个第三方比较器，用于比较两个对象。

在Scala中也引入了以上两种比较方法(Scala.math包下):

Ordered特质混入Java的Comparable接口，它定义了相同类型间的比较方式，但这种内部比较方式是单一的；

trait Ordered[A] extends Any with java.lang.Comparable[A]{......}

case class Project(tag:String, score:Int) extends Ordered[Project] { 
  def compare(pro:Project ) = tag.compareTo(pro.tag) 
}

object OrderedDemo { 
  def main(args: Array[String]): Unit = { 
    val list = List(Project("hadoop",60), Project("flink",90), Project("hive",70),Project("spark",80)) 
    println(list.sorted) 
  } 
}

Ordering特质混入Comparator接口，它是提供第三方比较器，可以自定义多种比较方式，在实际开发中也是使用比较多的，灵活解耦合。

trait Ordering[T] extends Comparator[T] with PartialOrdering[T] with Serializable {......}

object OrderingDemo { 
  def main(args: Array[String]): Unit = { 
    val pairs = Array(("a", 7, 2), ("c", 9, 1), ("b", 8, 3)) 
    // Ordering.by[(Int,Int,Double),Int](_._2)表示从Tuple3转到Int型 
    // 并按此Tuple3中第二个元素进行排序 
    Sorting.quickSort(pairs)(Ordering.by[(String, Int, Int), Int](_._2)) 
    println(pairs.toBuffer) 
  } 
}

模式匹配和样例类

模式匹配

模式匹配的基本语法结构：变量 match { case 值 => 代码 }

模式匹配match case中，只要有一个case分支满足并处理了，就不会继续判断下一个case分支了，不需要使用break语句。这点与Java不同，Java的switch case需要用break阻止。如果值为下划线，则代表不满足以上所有情况的时候如何处理。

模式匹配match case最基本的应用，就是对变量的值进行模式匹配。match是表达式，与if表达式一样，是有返回值的。

除此之外，Scala还提供了样例类，对模式匹配进行了优化，可以快速进行匹配。

匹配类型

Scala的模式匹配还有一个强大的功能，它可以直接匹配类型，而不是值。这一点是Java的switch case做不到的。

匹配类型的语法：变量 match {case 变量 : 类型 => **代码}**，而不是匹配值的“case 值 => 代码”这种语法。

匹配数组、元组、集合

def main(args: Array[String]): Unit = { 
  val arr = Array(0, 3, 5) 
  //对Array数组进行模式匹配，分别匹配： 
  //带有指定个数元素的数组、带有指定元素的数组、以某元素开头的数组 
  arr match { 
    case Array(0, x, y) => println(x + " " + y) 
    case Array(0) => println("only 0") 
    //匹配数组以1开始作为第一个元素 
    case Array(1, _*) => println("1 ...") 
    case _ => println("something else") 
  }
  
  val list = List(3, -1) 
  //对List列表进行模式匹配，与Array类似，但是需要使用List特有的::操作符 
  //构造List列表的两个基本单位是Nil和::，Nil表示为一个空列表 
  //tail返回一个除了第一元素之外的其他元素的列表 
  //分别匹配：带有指定个数元素的列表、带有指定元素的列表、以某元素开头的列表 
  list match { 
    case x :: y :: Nil => println(s"x: $x y: $y") 
    case 0 :: Nil => println("only 0") 
    case 1 :: tail => println("1 ...") 
    case _ => println("something else") 
  }
  
  val tuple = (1, 3, 7) 
  tuple match { 
    case (1, x, y) => println(s"1, $x , $y") 
    case (_, z, 5) => println(z) 
    case _ => println("else") 
  } 
}

样例类

case class样例类是Scala中特殊的类。当声明样例类时，以下事情会自动发生：

• 主构造函数接收的参数通常不需要显式使用var或val修饰，Scala会自动使用val修饰
• 自动为样例类定义了伴生对象，并提供apply方法，不用new关键字就能够构造出相应的对象
• 将生成toString、equals、hashCode和copy方法，除非显示的给出这些方法的定义
• 继承了Product和Serializable这两个特质，也就是说样例类可序列化和可应用Product的方法

case class是多例的，后面要跟构造参数，case object是单例的。

此外，case class样例类中可以添加方法和字段，并且可用于模式匹配。

class Amount 
//定义样例类Dollar，继承Amount父类 
case class Dollar(value: Double) extends Amount 

//定义样例类Currency，继承Amount父类 
case class Currency(value: Double, unit: String) extends Amount 

//定义样例对象Nothing，继承Amount父类
case object Nothing extends Amount 

object CaseClassDemo { 
  def main(args: Array[String]): Unit = { 
    judgeIdentity(Dollar(10.0)) 
    judgeIdentity(Currency(20.2,"100")) 
    judgeIdentity(Nothing) 
}

	//自定义方法，模式匹配判断amt类型 
	def judgeIdentity(amt: Amount): Unit = { 
  	amt match { 
    	case Dollar(value) => println(s"$value") 
    	case Currency(value, unit) => println(s"Oh noes,I got $unit") 
    	case Nothing => println("Oh,GOD!") 
  	} 
	} 
}

Option与模式匹配

object Option_Test {
  def main(args: Array[String]): Unit = {
    val grade = Map("a"->90,"b"->60,"c"->80)
    val temp: Int = grade("a")
    println(temp)
    // println(grade("q")) error
    def getGrade(name:String)={
      val maybeInt: Option[Int] = grade.get(name)
      maybeInt match {
        case None=>println("None")
        case Some(maybeInt)=>println(maybeInt)
      }
    }
    getGrade("b")
    getGrade("d")
  }
}

函数

基础

// 函数
val add1: Int => Int = (x) => x+1
// 方法
def add1(x: Int): Int = { x + 1 }
// 方法转函数
val add2 = add1 _

闭包

闭包是在其上下文中引用了自由变量的函数；

闭包引用到函数外面定义的变量，定义这个函数的过程就是将这个自由变量捕获而构成的一个封闭的函数，也可理解

为”把函数外部的一个自由变量关闭进来“。

何为闭包？需满足下面三个条件：

• 1、闭包是一个函数
• 2、函数必须要有返回值
• 3、返回值依赖声明在函数外部的一个或多个变量，用Java的话说，就是返回值和定义的全局变量有关

柯里化

函数编程中，接收多个参数的函数都可以转化为接收单个参数的函数，这个转化过程就叫柯里化(Currying)。

Scala中，柯里化函数的定义形式和普通函数类似，区别在于柯里化函数拥有多组参数列表，每组参数用小括号括起来。

Scala API中很多函数都是柯里化的形式。

// 使用普通的方式 
def add1(x: Int, y: Int) = x + y 

// 使用闭包的方式，将其中一个函数作为返回值 
def add2(x: Int) = (y:Int) => x + y 

// 使用柯里化的方式 
def add(x: Int)(y: Int) = x + y 

//调用柯里化函数add 
scala> add(1)(2) 
res1: Int = 3 

偏函数

偏函数（Partial Function）之所以“偏”，原因在于它们并不处理所有可能的输入，而只处理那些能与至少一个 case语句匹配的输入；

在偏函数中只能使用 case 语句，整个函数必须用大括号包围。这与普通的函数字面量不同，普通的函数字面量可以使用大括号，也可以用小括号；

被包裹在大括号中的一组case语句是一个偏函数，是一个并非对所有输入值都有定义的函数；

Scala中的Partial Function是一个trait，其类型为PartialFunction[A,B]，表示：接收一个类型为A的参数，返回一个类型为B的结果。

// 1、2、3有对应的输出值，其它输入打印 Other 
val pf: PartialFunction[Int, String] = { 
  case 1 => "One" 
  case 2 => "Two" 
  case 3 => "Three" 
  case _=> "Other" 
}

pf(1) // 返回: One 
pf(2) // 返回: Two 
pf(5) // 返回: Other

隐式转换

隐式转换函数

class Num {} 

class RichNum(num: Num) { 
  def rich(): Unit = { 
    println("Hello Implicit!") 
  } 
}

object ImplicitDemo { 
// 定义一个名称为num2RichNum的隐式函数 
	implicit def num2RichNum(num: Num): RichNum = { 
 		new RichNum(num) 
	}
  
  def main(args: Array[String]): Unit = { 
    val num = new Num 
    // num对象并没有rich方法，编译器会查找当前范围内是否有可转换的函数 
    // 如果没有则编译失败，如果有则会调用。 
    num.rich() 
  } 
}

泛型

泛型类

//定义一个泛型类 

class Stack[T1, T2, T3](name: T1) { 
  var age: T2 = _ 
  var address: T3 = _ 
  def getInfo: Unit = { 
    println(s"$name,$age,$address") 
  } 
}

泛型函数

object GenericityFunction { 
  def getCard[T](content: T) = { 
    content match { 
      case content: Int => s"card:$content is Int " 
      case content: String => s"card:$content is String" 
      case _ => s"card:$content" 
    } 
	}
  def main(args: Array[String]): Unit = { 
    println(getCard[String]("hello")) 
    println(getCard(1001)) 
  } 
}

协变和逆变

Scala的协变和逆变是非常有特色的，完全解决了Java中的泛型的一大缺憾！

举例来说，Java中，如果有Professional是Master的子类，那么Card[Professionnal]是不是Card[Master]的子类？

答案是：不是。因此对于开发程序造成了很多的麻烦。

而Scala中，只要灵活使用协变和逆变，就可以解决Java泛型的问题。

协变定义形式如：trait List[+T] {}

当类型S是类型A的子类型时，则List[S]也可以认为是List[A}的子类型，即List[S]可以泛化为List[A]，也就是被参数化，类型的泛化方向与参数类型的方向是一致的，所以称为协变（covariance）。

逆变定义形式如：trait List[-T] {}

当类型S是类型A的子类型，则Queue[A]反过来可以认为是Queue[S}的子类型，也就是被参数化类型的泛化方向与参数类型的方向是相反的，所以称为逆变（contravariance）。

 //大师 
class Master 

//专家 
class Professor extends Master 

//讲师class Teacher 
//这个是协变，Professor是Master的子类，此时Card[Profesor]也是Card[Master]的子类 
class Card[+T] 

object CovarianceDemo { 
  def enterMeet(card: Card[Master]): Unit = { 
    //只有Card[Master]及其子类Card[Professor]才能进入会场。 
    println("欢迎进入会场！") 
  }
  
  def main(args: Array[String]): Unit = { 
    val masterCard = new Card[Master] 
    val professorCard = new Card[Professor] 
    val teacharCard = new Card[Teacher] 
    enterMeet(masterCard) 
    enterMeet(professorCard) 
    //此处就会报错 
    // enterMeet(teacharCard) 
  } 
}

Spark基础

mapPartitions

与map方法类似，map是对rdd中的每一个元素进行操作，而mapPartitions(foreachPartition)则是对rdd中的每个分区的迭代器进行操作。如果在map过程中需要频繁创建额外的对象(例如将rdd中的数据通过jdbc写入数据库,map需要为每个元素创建一个链接而mapPartition为每个partition创建一个链接),则mapPartitions效率比map高的多。

/**
 * @Description: spark mapPartitions test
 * @Author: Xiaoyuyu
 * @CreateDate: 2021/4/22 1:41 下午
 */
object mapPartitions_test {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val value: RDD[Int] = sc.makeRDD((1 to 50).toArray)

    def fun1(iter:Iterator[Int]) : Iterator[(Int,Int)] = {
      val tuples = new ListBuffer[(Int, Int)]
      while(iter.hasNext) {
        val elem: Int = iter.next()
        val tuple: (Int, Int) = (elem, elem * 2)
        tuples+=tuple
      }
      tuples.toIterator
    }

    val value1: RDD[(Int, Int)] = value.mapPartitions(fun1)
    value1.foreach(println)

    sc.stop()
  }
}

本文作者： xiaoyuyu
本文链接： http://woaixiaoyuyu.github.io/2021/07/03/Spark%E5%B8%B8%E7%94%A8%E4%BB%A3%E7%A0%81%E6%AE%B5/
版权声明： 本作品采用 知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议 进行许可。转载请注明出处！

赏

谢谢你请我吃糖果