[DB] Spark Core (3)

高级算子

• mapPartitionWithIndex：对RDD中每个分区（有下标）进行操作，通过自己定义的一个函数来处理
  • def mapPartitionsWithIndex[U](f: (Int, Iterator[T]) ⇒ Iterator[U])
  • f 是函数参数，接收两个参数
    • Int：分区号
    • Iterator[T]：分区中的元素
  • Iterator[U]：处理完的结果

• aggregate：聚合操作（类似分组）
  • 先对局部进行聚合操作，再对全局进行聚合操作
  • rdd1.aggregate(0)(max(_,_),_+_) 结果 7
  • rdd1.aggregate(10)(max(_,_),_+_) 结果 30

• aggregateByKey：类似aggregate，操作<Key Value>
• coalesce：重分区，默认不会进行shuffle
• repartition：重分区，对数据进行shuffle

编程案例

• 分析Tomcat的访问日志，找到访问最高的两个网页
  • 对网页访问量求和
  • 排序（降序）

package day0608

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

object MyTomcatLogCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("MyTomcatLogCount").setMaster("local")
    val sc = new SparkContext(conf)

    /*
     * 日志：192.168.88.1 - - [30/Jul/2017:12:53:43 +0800] "GET /MyDemoWeb/head.jsp HTTP/1.1" 200 713
     * 返回：(hadoop.jsp,1)，相当于WordCount中的<k2 v2>
     */
    val rdd1 = sc.textFile("G:\K\TZ-Bigdata\讲义\1101-Spark案例分析\代码\localhost_access_log.2017-07-30.txt").map(line => {
      //解析字符串，找到jsp的名字
      //得到双引号位置
      val index1 = line.indexOf(""")
      val index2 = line.lastIndexOf(""")
      val line1 = line.substring(index1+1,index2) // 得到 GET /MyDemoWeb/head.jsp HTTP/1.1

      //得到两个空格位置
      val index3 = line1.indexOf(" ")
      val index4 = line1.lastIndexOf(" ")
      val line2 = line1.substring(index3+1,index4) // 得到 /MyDemoWeb/head.jsp

      //得到jsp的名字
      val jspName = line2.substring(line2.lastIndexOf("/") + 1)

      //返回
      (jspName,1)
    })
    //按照jsp的名字进行聚合操作，类似WordCount
    val rdd2 = rdd1.reduceByKey(_+_) // 得到所有jsp访问总量，如(hadoop.jsp,9) (oracle.jsp,9)

    //排序，按value降序顺序
    val rdd3 = rdd2.sortBy(_._2,false)

    //取出访问量最高的两个网页
    println(rdd3.take(2).toBuffer)
    sc.stop()
  }
}

ArrayBuffer((oracle.jsp,9), (hadoop.jsp,9))

• 创建自定义分区

package day0608

import org.apache.spark.{Partitioner, SparkConf, SparkContext}
import scala.collection.mutable.HashMap

object MyTomcatLogPartitioner {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("MyTomcatLogPartitioner").setMaster("local")
    val sc = new SparkContext(conf)

    /*
     * 日志：192.168.88.1 - - [30/Jul/2017:12:53:43 +0800] "GET /MyDemoWeb/head.jsp HTTP/1.1" 200 713
     * 返回：(hadoop.jsp,对应的日志)，相当于WordCount中的<k2 v2>
     */

    val rdd1 = sc.textFile("G:\K\TZ-Bigdata\讲义\1101-Spark案例分析\代码\localhost_access_log.2017-07-30.txt")
      .map(line => {
      //解析字符串，找到jsp的名字
      //得到双引号位置
      val index1 = line.indexOf(""")
      val index2 = line.lastIndexOf(""")
      val line1 = line.substring(index1+1,index2) // 得到 GET /MyDemoWeb/head.jsp HTTP/1.1

      //得到两个空格位置
      val index3 = line1.indexOf(" ")
      val index4 = line1.lastIndexOf(" ")
      val line2 = line1.substring(index3+1,index4) // 得到 /MyDemoWeb/head.jsp

      //得到jsp的名字
      val jspName = line2.substring(line2.lastIndexOf("/") + 1)

      //返回(jsp的名字，访问日志)
      (jspName,line)
    })

    //得到不重复的jsp名字，创建分区规则
    val rdd2 = rdd1.map(_._1).distinct().collect()

    //创建分区规则
    val myPartitioner = new MyWebPartitioner(rdd2)

    //对rdd1进行分区
    val rdd3 = rdd1.partitionBy(myPartitioner)

    //输出
    rdd3.saveAsTextFile("G:\K\TZ-Bigdata\讲义\1101-Spark案例分析\output")

    sc.stop()
  }
}

//根据jsp名字，创建分区规则
class MyWebPartitioner(jspList:Array[String]) extends Partitioner{
  //定义集合保存分区条件
  //String：jsp的名字 Int：对应的分区号
  val partitionMap = new HashMap[String,Int]()

  var partID = 0
  for(jsp <- jspList){
    partitionMap.put(jsp,partID)
    partID += 1
  }
  //实现抽象方法
  //返回有多少分区
  override def numPartitions:Int = partitionMap.size

  //根据jsp的名字key，查找对应的分区号
  override def getPartition(key: Any):Int = {
    partitionMap.getOrElse(key.toString,0)
  }
}

   

• 操作数据库（把结果存入MySQL）
  • 对分区进行操作
  • conn和pst在不同的分区（节点）上进行使用

package day0611

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import java.sql.DriverManager

object MyTomcatLogCountToMysql {

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

    val conf = new SparkConf().setMaster("local").setAppName("MyTomcatLogCountToMysql")
    val sc = new SparkContext(conf)

    val rdd1 = sc.textFile("G:\K\TZ-Bigdata\讲义\1101-Spark案例分析\代码\localhost_access_log.2017-07-30.txt")
      .map(
        line => {
          //解析字符串 找到jsp名字
          //得到两个双引号之间的东西  GET /MyDemoWeb/hadoop.jsp HTTP/1.1
          val index1 = line.indexOf(""")
          val index2 = line.lastIndexOf(""")
          val line1 = line.substring(index1 + 1, index2)
          //得到两个空格之间的东西 /MyDemoWeb/hadoop.jsp
          val index3 = line1.indexOf(" ")
          val index4 = line1.lastIndexOf(" ")
          val line2 = line1.substring(index3 + 1, index4)
          //得到jsp名字
          val jspName = line2.substring(line2.lastIndexOf("/") + 1)

          (jspName, 1)
        })

    //        var conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/company?serverTimezone=UTC&characterEncoding=utf-8", "root", "123456")
    //        var pst = conn.prepareStatement("insert into mydata values(?,?)")
    //
    //        rdd1.foreach(f => {
    //        pst.setString(1, f._1)
    //        pst.setInt(2,f._2)
    //        pst.executeUpdate()
    //     })

    //  上述代码直接运行时报错：Task not serializable
    //  因为 conn 和 pst 没有序列化 即 不能再不同节点上进行传输
    //

    //    rdd1.foreach(f => {
    //      var conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/company?serverTimezone=UTC&characterEncoding=utf-8", "root", "Chen1227+")
    //      var pst = conn.prepareStatement("insert into mydata values(?,?)")
    //      pst.setString(1, f._1)
    //      pst.setInt(2, f._2)
    //      pst.executeUpdate()
    //    })


    //     上述代码可直接运行 相当于在本地新建连接
    //     每条数据都创建Connection和PreparedStatement
    //     缺点：频繁操作数据库 对数据库压力很大


    //第二种修改方式，针对分区进行操作，每个分区创建一个conn 和 pst
    //参数要求 (f: Iterator[(String, Int)] => Unit): Unit
    //相当于 对 rdd1 中每个分区都调用 saveToMysql 函数
    rdd1.foreachPartition(saveToMysql)
    sc.stop()
  }
    //
    //  }

    //  //定义一个函数 针对分区进行操作
      def saveToMysql(it: Iterator[(String, Int)]) {
        //it保存的是一个分区的数据
        var conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/company?serverTimezone=UTC&characterEncoding=utf-8", "root", "Chen1227+")
        var pst = conn.prepareStatement("insert into mydata values(?,?)")

        it.foreach(f => {
          pst.setString(1, f._1)
          pst.setInt(2, f._2)
          pst.executeUpdate()
        })
      }

}

   

参考

RDD算子文档

http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html