Spark性能调优

配置：
3台 48核 64GB
具体调参如下：
考虑Linux运行及程序、Hadoop、Yarn等守护进程等，约占5%-10%，每台预留3核心以及4G内存。

为每个执行器分配3个核心
--executor-cores = 3
 
每个节点除去预留核心，剩下： 48-3 = 45
群集中核心的可用总数： 45 x 3 = 135
 
–-num-executors = 群集中核心的可用总数/每个executors分配3核心数
                = 135/3 
                = 45
每个节点的executors数目： 45/3 = 15

群集中每个节点的可使用的总内存数： 64GB - 4GB = 60GB
 
--executor-memory = 每个executor的内存= 60GB / 15 
                  = 4GB
 
预留的 off heap overhead = 4GB * Max(384MB, 7% of 4GB)
--executor-memory = 4 - 384M 
                  ≈ 4GB
PS:经参数调优以及测试之后，此为最佳num-executors、executor-memory、executor-core数目设置

object Demo1Cache {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    val studentsRDD: RDD[String] = sc.textFile("data/students.txt")
    /**
     * 当对同一个rdd进行多次使用的时候可以将rdd缓存起来
     *
     */
    //缓存级别是MEMORY_ONLY
    //studentsRDD.cache()
    //内存放不下放磁盘，同时会对数据做序列化，将一个分区的数据序列化从一个字节数组
    studentsRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)
    /**
     * rdd: rdd.cache
     * df : df.cache
     * sql: cache table student,  uncache table studnet
     */
    /**
     * 统计班级的的人数
     *
     */
    studentsRDD
      .map(stu => (stu.split(",")(3), 1))
      .reduceByKey(_ + _)
      .map {
        case (clazz: String, num: Int) =>
          s"$clazzt$num"
      }
      .saveAsTextFile("data/cache/clazz_num")
    /**
     * 统计性别的人数
     *
     */
    studentsRDD
      .map(stu => (stu.split(",")(3), 1))
      .reduceByKey(_ + _)
      .map {
        case (gender: String, num: Int) =>
          s"$gendert$num"
      }
      .saveAsTextFile("data/cache/gender_num")
    /**
     * 清空缓存
     */
    studentsRDD.unpersist()
    while (true) {
    }
  }
}

1 使用reduceByKey/aggregateByKey替代groupByKey
实例如下：
object Demo2AggregateByKey{
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("Demo2AggregateByKey")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    val studentsRDD: RDD[String] = sc.textFile("data/students.txt")
    val clazzKvDS: RDD[(String, Int)] = studentsRDD.map(stu => (stu.split(",")(4), 1))
    /**
     * aggregateByKey: 需要两个函数，一个是map端预聚合的函数，一个reduce端汇总的函数
     * reduceByKey map端和reduce端聚合函数是一样，
     * 如果map端和reduce端要写不一样的聚合函数可以使用aggregateByKey
     */
    val countRDD: RDD[(String, Int)] = clazzKvDS.aggregateByKey(0)(
      (u: Int, i: Int) => u + i,//在map端做聚合函数
      (u1: Int, u2: Int) => u1 + u2//在reduce端做聚合的函数
    )
    countRDD.foreach(println)
  }
}
2 使用mapPartitions替代普通map Transformation算子
实例如下：
object Demo3MapPartition {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    val dataRDD: RDD[String] = sc.textFile("data/students.txt")
    val kvRDD: RDD[(String, String, String)] = dataRDD.mapPartitions(iter => {
      iter.map(line => {
        //如果只是简单的拆分数据，使用map和mappartition没有区别
        val split: Array[String] = line.split("t")
        (split(0), split(1), split(2))
      })
    })
    val resultRDD: RDD[(String, Long, String)] = kvRDD.mapPartitions(iter => {
      /**
       *
       * 可以将一些初始化的代码放在mapPartitions中，减少占用的内存空间
       */
      //将时间字段转换成时间戳
      //在这里创建的对象，是一个分区创建一个
      val format = new SimpleDateFormat("yyyy/MM/dd")
      iter.map {
        case (id: String, sdate: String, p: String) =>
          val dateObj: Date = format.parse(sdate)
          val ts: Long = dateObj.getTime
          (id, ts, p)
      }
    })
    resultRDD.foreach(println)
  }
}
3 使用foreachPartitions替代foreach Action算子
实例如下：
object Demo4ForeachPartition {
    def main(args: Array[String]): Unit = {
      val spark: SparkSession = SparkSession.builder()
        .master("local")
        .appName("foreach")
        .getOrCreate()
      val rdd1: RDD[String] = spark
        .sparkContext
        .textFile("data/student.txt", 4)
      println(rdd1.getNumPartitions)
      /*
        * foreachPartition: 遍历一个分区的数据
        * 如果需要将数据保存到外部数据库，使用foreachPartition代替foreach
        * foreachPartition 每一个分区只会创建一个连接
        */
      rdd1.foreachPartition(iter => {
        //这里的代码每一个分区只会执行一次
        //每一个分区只会建立一个连接
        Class.forName("com.mysql.jdbc.Driver")
        val con: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/student?useUnicode=true&characterEncoding=utf-8", "root", "123456")
        println("连接建立成功")
        //遍历一个分区的数据
        iter.foreach(line => {
          val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")
          val split: Array[String] = line.split(",")
          stat.setString(1, split(0))
          stat.setString(2, split(1))
          stat.setInt(3, split(2).toInt)
          stat.setString(4, split(3))
          stat.setString(5, split(4))
          //插入数据
          stat.executeUpdate()
        })
        //关闭连接
        con.close()
      })
    }
}
4 使用filter之后进行coalesce操作
5 repartition:coalesce(numPartitions，true)增多分区使用这个
6 coalesce(numPartitions，false)减少分区，没有shuffle只是合并partition
4 5 6实例如下：
object Demo5FilterCoalesceRepartition {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)
    val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6,7,8,9,10,11,12),3)
    val result = rdd.filter(
      item => item % 2 == 0
    )
    result.collect().foreach(println)
    print("-------------------以上为filter结果，以下为先filter后coalesce结果----------------------------------")
    // 缩减分区：coalesce，如果想要数据均衡，可以采用shuffle
    val coalesce_result = result.coalesce(2,true)
    coalesce_result.collect().foreach(println)
    print("--------------------以上为先filter后coalesce结果，以下为先filter后repartition结果-------------------")
    // 扩大分区：repartition, 底层代码调用的就是coalesce，而且肯定采用shuffle
    val repartition_result = result.repartition(6)
    repartition_result.collect().foreach(println)
    print("--------------------以上为先filter后repartition结果，以下为coalesce(numPartitions，false)减少分区结果")
    // coalesce(numPartitions，false) 减少分区，没有shuffle只是合并partition
    val coalesce_falseresult = result.coalesce(2,false)
    coalesce_falseresult.collect().foreach(println)
    sc.stop()
  }
}
7 使用repartitionAndSortWithinPartitions替代repartition与sort类操作代码
实例如下：
//创建key类，key组合键为grade，score
case class StudentKey(grade:String,score:Int)
object StudentKey {
  implicit def orderingByGradeStudentScore[A <: studentkey : ordering>    Ordering.by(fk => (fk.grade, fk.score * -1))
  }
}
object Demo6Student{
  def main(args: Array[String]) {
    //定义hdfs文件索引值
    val grade_idx:Int=0
    val student_idx:Int=1
    val course_idx:Int=2
    val score_idx:Int=3
    //定义转化函数，不能转化为Int类型的，给默认值0
    def safeInt(s: String): Int = try { s.toInt } catch { case _: Throwable  => 0 }
    //定义提取key的函数
    def createKey(data: Array[String]):StudentKey={
      StudentKey(data(grade_idx),safeInt(data(score_idx)))
    }
    //定义提取value的函数
    def listData(data: Array[String]):List[String]={
      List(data(grade_idx),data(student_idx),data(course_idx),data(score_idx))
    }
    def createKeyValueTuple(data: Array[String]) :(StudentKey,List[String]) = {
      (createKey(data),listData(data))
    }
    //创建分区类
    import org.apache.spark.Partitioner
    class StudentPartitioner(partitions: Int) extends Partitioner {
      require(partitions >= 0, s"Number of partitions ($partitions) cannot be negative.")
      override def numPartitions: Int = partitions
      override def getPartition(key: Any): Int = {
        val k = key.asInstanceOf[StudentKey]
        k.grade.hashCode() % numPartitions
      }
    }
    //设置master为local，用来进行本地调试
    val conf = new SparkConf().setAppName("Student_par
tition_sort").setMaster("local")
    val sc = new SparkContext(conf)
    //随机数据
    val student_array =Array(
      "c001,n003,chinese,49",
      "c002,n004,english,79",
      "c002,n004,chinese,13",
      "c001,n001,english,88",
      "c001,n002,chinese,10",
      "c002,n006,chinese,29",
      "c001,n001,chinese,54",
      "c001,n002,english,32",
      "c001,n003,english,43",
      "c002,n005,english,80",
      "c002,n005,chinese,48",
      "c002,n006,english,69"
    )
    //将学生信息并行化为rdd
    val student_rdd = sc.parallelize(student_array)
    //生成key-value格式的rdd
    val student_rdd2 = student_rdd.map(line => line.split(",")).map(createKeyValueTuple)
    //根据StudentKey中的grade进行分区，并根据score降序排列
    val student_rdd3 = student_rdd2.repartitionAndSortWithinPartitions(new StudentPartitioner(10))
    //打印数据
    student_rdd3.collect.foreach(println)
  }
}

object Demo7Mapjoin {
  /**
    * map join将小表广播，大表使用map算子
    * 1、小表不能太大,不能超过2G
    * 2、如果driver内存不足，需要手动设置;如果广播变量大小超过了driver内存大小，会出现oom
    */
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("app")
    val sc: SparkContext = new SparkContext(conf)
    //RDD 不能广播
    val studentRDD: RDD[String] = sc.textFile("data/students.txt")
    //将数据拉去到driver端，变成一个map集合
    val stuMap: Map[String, String] = studentRDD
      .collect() //将rdd的数据拉取Driver端变成一个数组
      .map(s => (s.split(",")(0), s))
      .toMap
    //广播map集合
    val broStu: Broadcast[Map[String, String]] = sc.broadcast(stuMap)
    val scoreRDD: RDD[String] = sc.textFile("data/students.txt")
    //循环大表，通过key获取小表信息
    scoreRDD.map(s => {
      val sId: String = s.split(",")(0)
      //重广播变量里面获取数据
      val stuInfo: String = broStu.value.getOrElse(sId, "")
      stuInfo + "," + s
    }).foreach(println)
    while (true) {
    }
  }
}

第一步，创建SparkConf对象。
val conf = new SparkConf().setMaster(...).setAppName(...)
第二步，在SparkConf中设置序列化器为KryoSerializer。
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") 
第三步，注册你使用的需要通过Kryo序列化的一些自定义类，SparkConf.registerKryoClasses()。 
项目中的使用： 
conf.registerKryoClasses(new Class[]{CategorySortKey.class})