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)
#循环打印结果
})
}
}