亲宝软件园·资讯

展开

Spark学习进度7-综合案例

清风紫雪 人气: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.转换格式如下

 

 

 可用图片展示:

 

 

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(_))

  }

连接操作

案例介绍:

有两个表: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表:

 

 最终输出:

 

 

 

加载全部内容

相关教程
猜你喜欢
用户评论