微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦!

spark第六次课

import org.apache.spark.{SparkConf, SparkContext}
#导入需要的类

object TopN {

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

    val conf = new SparkConf().setAppName("TopN").setMaster("local")
    #设置一些配置

    val sc = new SparkContext(conf)
    #连接SparkMaster,生成sc对象

    sc.setLogLevel("ERROR")
     #设置日志输出格式

    val lines = sc.textFile("hdfs://localhost:9000/user/hadoop/spark/mycode/rdd/examples",2)
    #读取文本文件所有行的内容生成一个RDD,即lines(每个元素都是字符串,是文本中的一行)
    #2为设置2个分区

    var num = 0;
    #显示结果时,前面的序号

    val result = lines.filter(line => (line.trim().length > 0) && (line.split(",").length == 4))
    #将空行和字段数量不等于4的行丢弃掉(假设当前为rdd1)

      .map(_.split(",")(2))
    #将rdd1的每个元素拆分成四个字符串,保存到数组中,并把数组的第三个元素取出放到新的RDD中(假设当前为rdd2)

      .map(x => (x.toInt,""))
      #将rdd2的每个元素从String转换成Int,并生成(x,"")键值对的形式放到新的RDD中(假设当前为rdd3)

      .sortByKey(false)
      #将rdd3降序排序(假设当前为rdd4)

      .map(x => x._1).take(5)
      #取出rdd4前五的key值(假设当前为rdd5)

      .foreach(x => {

        num = num + 1

        println(num + "\t" + x)
        #循环打印结果

      })

  }

}

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 [email protected] 举报,一经查实,本站将立刻删除。

相关推荐