日志类型

测试数据

http://bigdata.myit.com/zhangsan

http://bigdata.myit.com/zhangsan

http://bigdata.myit.com/zhangsan

http://bigdata.myit.com/zhangsan

http://bigdata.myit.com/zhangsan

http://java.myit.com/lisi

http://java.myit.com/lisi

http://java.myit.com/lisi

计算每个学科最受欢迎的老师

package mypro

import java.net.URL

import org.apache.log4j.{Level, Logger}

import org.apache.spark.rdd.RDD

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

/**

 * Created by 166 on 2017/9/5.

 */

object FavTeacher {

  def main(args: Array[String]) {

    Logger.getLogger("org.apache.spark").setLevel(Level.OFF)

    val conf = new SparkConf().setAppName(this.getClass.getName).setMaster("local[2]")//local[*]代表用多个线程跑,2代表用两个线程跑

    val sc = new SparkContext(conf)

    //读取数据

    val lines: RDD[String] = sc.textFile(args())

    //整理数据

    val subjectAndTeacher:RDD[(String,String)]=lines.map(line=> {

      val url = new URL(line)

      val host = url.getHost

      val subject = host.substring(, host.indexOf("."))

      val teacher = url.getPath.substring()   //去掉路径前面的"/"

      (subject, teacher)

    })

    //聚合

    val reduce = subjectAndTeacher.map((_,)).reduceByKey(_+_)

    //println(reduce.collect().toBuffer)

    //按学科分组

    val grouped: RDD[(String, Iterable[((String, String), Int)])] = reduce.groupBy(_._1._1)//迭代器不能排序,需要将它变成List。

    //二次排序

    val result: RDD[(String, List[((String, String), Int)])] = grouped.mapValues(_.toList.sortBy(_._2).reverse.take())//用scala的语法，会把数据全部加载到内存后再做排序，数据量大的时候会有性能问题，内存溢出的问题，不建议这样使用，

    val arr: Array[(String, List[((String, String), Int)])] = result.collect()

    println(arr.toBuffer)

  }

}

另种角度来实现，过滤多次提交

package com.rz.spark.base

import java.net.URL

import org.apache.log4j.{Level, Logger}

import org.apache.spark.rdd.RDD

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

// 过滤多次提交

object GroupFavTeacher2 {

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

    Logger.getLogger("org.apache.spark").setLevel(Level.OFF)

    val conf = new SparkConf().setMaster("local[2]").setAppName(this.getClass.getSimpleName)

    val sc = new SparkContext(conf)

    val topN = args(1).toInt

    val subject = Array("bigdata","javaee","php")

    // 读取数据

    val lines: RDD[String] = sc.textFile(args(0))

    // 整理数据  http://bigdata.myit.cn/laozhang

    val subjectAndTeacher= lines.map(line => {

      val url = new URL(line)

      val host = url.getHost

      val subject = host.substring(0, host.indexOf("."))

      val teacher = url.getPath.substring(1) // 去掉前面的/

      ((subject, teacher),1)

    })

    // 聚合

    val reduced = subjectAndTeacher.reduceByKey(_+_)

　　
　　// 缓存到内存,因为多次过滤都是使用同一个rdd，缓存到内存可以提高反复使用的性能
　　val cache = reduced.cache()

    for (sb <- subject){

      val sorted = cache.filter(_._1._1 == sb).sortBy(_._2,false).take(topN)

      println(sorted.toBuffer)

    }

    sc.stop()

  }

}

　　使用自定义分区器将每个学科的数据shuffle到独自的分区，在分区内进行排序取topN

package com.rz.spark.base

import java.net.URL

import org.apache.log4j.{Level, Logger}

import org.apache.spark.rdd.RDD

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

// 自定义分区器

import scala.collection.mutable

// 过滤多次提交

object GroupFavTeacher3 {

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

    Logger.getLogger("org.apache.spark").setLevel(Level.OFF)

    val conf = new SparkConf().setMaster("local[2]").setAppName(this.getClass.getSimpleName)

    val sc = new SparkContext(conf)

    val topN = args().toInt

    val subject = Array("bigdata","javaee","php")

    // 读取数据

    val lines: RDD[String] = sc.textFile(args())

    // 整理数据  http://bigdata.myit.cn/laozhang

    val subjectAndTeacher= lines.map(line => {

      val url = new URL(line)

      val host = url.getHost

      val subject = host.substring(, host.indexOf("."))

      val teacher = url.getPath.substring() // 去掉前面的/

      ((subject, teacher),)

    })

    // 聚合

    val reduced = subjectAndTeacher.reduceByKey(_+_)

    // 计算我们有多少学科

    val sujects: Array[String] = reduced.map(_._1._1).distinct().collect()

    // 自定义一个分区器，并且按照指定的分区器进行分区

    val subjectPartitoner = new SubjectPartitoner(sujects)

    // partitionBy按照指定的分区规则进行分区

    val partitioned: RDD[((String, String), Int)] = reduced.partitionBy(subjectPartitoner)

    // 如果一次拿出一个分区（可以操作一个分区的数据）

    val sorted = partitioned.mapPartitions(it => {

      // 将迭代器转成List，然后排序，再转成迭代器返回

      it.toList.sortBy(_._2).reverse.take(topN).toIterator // 按数值排序

    })

    val result = sorted.collect()

    println(result.toBuffer)

    sc.stop()

  }

  // 自定义分区器

  class SubjectPartitoner(sbs: Array[String]) extends Partitioner{

    // 相当于主构造器(new 的时候会执行一次)

    // 用于存放规则的一个map

    val rules = new mutable.HashMap[String, Int]()

    var i =

    for (sb <- sbs){

      rules.put(sb,i)

      i +=

    }

    // 返回分区的数量（下一个RDD有多少分区）

    override def numPartitions: Int = sbs.length

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

    // Key是一个无组（String, String）

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

      // 获取学科名称

      val subject = key.asInstanceOf[(String, String)]._1

      // 根据规则计算分区编号

      rules(subject)

    }

  }

}

上面的方式会有多次shuffle，reduceByKey聚合数据的时候shuffle一次，使用自定义分区器重新对数据进行分析又shuffle了一次。我们可以尽可能的减少shuffle的过程，我们可以在reduceByKey的时候手动使用自定分区器进行分区，reduceByKey默认使用的是。HashPartitioner。

package com.rz.spark.base

import java.net.URL

import org.apache.log4j.{Level, Logger}

import org.apache.spark.rdd.RDD

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

// 自定义分区器且减少shuffle

import scala.collection.mutable

// 过滤多次提交

object GroupFavTeacher4 {

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

    Logger.getLogger("org.apache.spark").setLevel(Level.OFF)

    val conf = new SparkConf().setMaster("local[2]").setAppName(this.getClass.getSimpleName)

    val sc = new SparkContext(conf)

    val topN = args().toInt

    val subject = Array("bigdata","javaee","php")

    // 读取数据

    val lines: RDD[String] = sc.textFile(args())

    // 整理数据  http://bigdata.myit.cn/laozhang

    val subjectAndTeacher= lines.map(line => {

      val url = new URL(line)

      val host = url.getHost

      val subject = host.substring(, host.indexOf("."))

      val teacher = url.getPath.substring() // 去掉前面的/

      ((subject, teacher),)

    })

    // 计算我们有多少学科

    val sujects: Array[String] = subjectAndTeacher.map(_._1._1).distinct().collect()

    // 自定义一个分区器，并且按照指定的分区器进行分区

    val subjectPartitoner = new SubjectPartitoner2(sujects)

    // 聚合,聚合是按照指定的分区器进行分区

    // 该RDD一个分区内仅有一个学科的数据

    val reduced: RDD[((String, String), Int)] = subjectAndTeacher.reduceByKey(subjectPartitoner,_+_)

    // 如果一次拿出一个分区（可以操作一个分区的数据）

    val sorted = reduced.mapPartitions(it => {

      // 将迭代器转成List，然后排序，再转成迭代器返回

      it.toList.sortBy(_._2).reverse.take(topN).toIterator // 按数值排序

    })

    // 收集数据

    val result = sorted.collect()

    println(result.toBuffer)

    sc.stop()

  }

  // 自定义分区器

  class SubjectPartitoner2(sbs: Array[String]) extends Partitioner{

    // 相当于主构造器(new 的时候会执行一次)

    // 用于存放规则的一个map

    val rules = new mutable.HashMap[String, Int]()

    var i =

    for (sb <- sbs){

      rules.put(sb,i)

      i +=

    }

    // 返回分区的数量（下一个RDD有多少分区）

    override def numPartitions: Int = sbs.length

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

    // Key是一个无组（String, String）

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

      // 获取学科名称

      val subject = key.asInstanceOf[(String, String)]._1

      // 根据规则计算分区编号

      rules(subject)

    }

  }

}

最新文章

  1. List接口、Set接口、Map接口的方法
  2. [Java Basics] Reflection
  3. EMR,电子病历（Electronic Medical Record）
  4. 线段树的区间更新---A Simple Problem with Integers
  5. VSTO安装部署(完美解决XP+2007)
  6. systemd service
  7. Android常用的一些make命令（转载）--不错
  8. 用Delphi进行word开发
  9. Swift - 使用ALAssetsLibrary获取相簿里所有图片，视频（附样例）
  10. Azure Powershell对ARM资源的基本操作
  11. Rx响应式编程
  12. Vue技术内幕 出去看看吧 榨干部分小细节
  13. MySql CURD操作（数据的增删改查）
  14. 关于git 指令
  15. 转：自旋锁（spinlock）
  16. 一文理解 Java NIO 核心组件
  17. Powershell script to install Windows Updates (msu) from folder
  18. 《SQL Server 2008从入门到精通》--20180724
  19. P2513 [HAOI2009]逆序对数列
  20. AngularJs -- 指令中使用子作用域

热门文章

  1. 内核initcall分析
  2. Swift开发图解入门
  3. linux 参数内核
  4. laravel学习之路3 数据库相关
  5. Mina代码跟踪（1）
  6. Linux进程间通信(一) - 管道
  7. drupal7 使用（hook_preprocess_HOOK）向各个主题模版里面传递变量
  8. 【BZOJ2081】[Poi2010]Beads hash+调和级数
  9. rm -rf 删除文件找回
  10. iOS 蓝牙开发之(CoreBlueTooth)