spark学习七

综合案例

1 综合案例

1.0 文件排序

 解法：

1.读取数据

2.数据清洗，变换数据格式

3.从新分区成一个分区

4.按照key排序，返还带有位次的元组

5.输出

@Test
  def filesort(): Unit ={
    val source=sc.textFile("dataset/filesort.txt",3)
    var index=0
    /*
    partitionBy:把所有的分区相关的数据组成一个新的分区
    HashPartitioner(1)：分成一个分区，使得在一个分区内总体有序
     */
    val result= source.filter(_.trim().length>0).map(n => (n.trim.toInt,""))
      .partitionBy(new HashPartitioner(1))
      .sortByKey().map( t=> {
      index+=1
      (index,t._1)
    })
    result.foreach(println(_))
  }

1.2 二次排序

 

题目大意：先按照第一个比，相同则按照第二个比

题意思路：

1.读取数据

2.转换格式如下

 

 可用图片展示：

class SecondarySortKey(val first:Int,val second:Int) extends Ordered
  [SecondarySortKey] with Serializable{
  override def compare(that: SecondarySortKey): Int = {
    if(this.first-that.first!=0){
      this.first-that.first
    }else {
      this.second-that.second
    }
  }
}

//二次排序
  @Test
  def sortsecond(): Unit ={
    val source=sc.textFile("dataset/secondsort.txt",3)
    val secondrdd = source.map(item => (new SecondarySortKey(item.split(" ")(0).toInt, item.split(" ")(1).toInt), item))
        .partitionBy(new HashPartitioner(1))
    secondrdd.sortByKey(false)
      .map(item => item._2)
      .foreach(println(_))
  }

1.3 连接操作

案例介绍：

有两个表：movie表，和score表

score:包含的信息为：用户ID，电影ID,电影评分

movie:电影ID，电影名字

 我们想要得到，评分超过4分的（电影ID，电影名字，电影评分）

思路如下：

首先先弄score表：

1.获取想要的信息

2.获取对应电影ID的平均值

3.更换格式：keyBy，如下

对于movie表进行连接，连接前需要变化下格式

 然后可通过相同的key进行连接join，后的结果如下：

 进行评分的过滤，然后取出需要的数据

@Test
  /*
  score:包含的信息为：用户ID，电影ID,电影评分
  movie:电影ID，电影名字
   */
  def joinTest(): Unit ={
    val scoreRDD=sc.textFile("dataset/score.txt")
      .map(line => {
        val filed=line.split(",")
        (filed(1).toInt,filed(2).toDouble)
      })
      .groupByKey()
      .map(data =>{
        val avg=data._2.sum/data._2.size
        (data._1,avg)
      })
      .keyBy(it =>it._1)
    val movie=sc.textFile("dataset/movie.txt")
      .map(line => {
        val filed=line.split(",")
        (filed(0).toInt,filed(1))
      })
      .keyBy(it =>it._1)
    scoreRDD.join(movie)
      .filter(item => item._2._1._2>4.0)
      .map(it => (it._1,it._2._2._2,it._2._1._2))
      .foreach(println(_))
  }

输出：

score表：

 movie表：

 最终输出：