spark学习进度07（深入了解RDD）

package cn.itcast.spark.rdd

import org.junit.Test
import org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}

import scala.tools.scalap.scalax.util.StringUtil

/**
  * @Author 带上我快跑
  * @Data 2021/1/8 20:49
  * @菩-萨-说-我-写-的-都-对@
  */
class AccessLogAgg {

  @Test
  def ipAgg():Unit={

    //1、创建sparkcontext
    val conf=new SparkConf().setMaster("local[6]").setAppName("ip_agg")
    val sc=new SparkContext(conf)
    //2、读入文件生成数据集
    val sourceRDD=sc.textFile("dataset/access_log_sample.txt")
    //3、取出ip,赋予抽象次数为一
    val ipRDD=sourceRDD.map(item=>(item.split(" ")(0),1))
    //4、简单的清洗(去掉空数据，去掉非法的数据，根据业务在规整一下数据)
    val cleanRDD=ipRDD.filter(item => StringUtils.isNotEmpty(item._1))
    //5、根据ip出现的次数进行聚合
    val ipAggRDD=cleanRDD.reduceByKey((curr,agg)=>curr+agg)
    //6、根据ip出现的次数进行排序
    val sortedRDD=ipAggRDD.sortBy(item=>item._2,ascending = false)
    //7、取出结果，打印结果
    sortedRDD.take(10).foreach(item=>println(item))
    //8、关闭sc
    sc.stop()
  }
}