原始數(shù)據(jù)分析 0067011990999991950051507004888888889999999N9+00001+9999999999999999999999 0067011990999991950051512004888888889999999N9+00221+9999999999999999999999 0067011990999991950051518004888888889999999N9-00111+9999999999999999999999 0067011990999991949032412004888888889999999N9+01111+9999999999999999999999 0067011990999991950032418004888888880500001N9+00001+9999999999999999999999 0067011990999991950051507004888888880500001N9+00781+9999999999999999999999
數(shù)據(jù)說(shuō)明： 第15-19個(gè)字符是year 第45-50位是溫度表示，+表示零上 -表示零下，且溫度的值不能是9999，9999表示異常數(shù)據(jù) 第50位值只能是0、1、4、5、9幾個(gè)數(shù)字
package ClassicCaseimport org.apache.spark.{SparkConf, SparkContext}/** * 業(yè)務(wù)場(chǎng)景：通過(guò)采集的氣象數(shù)據(jù)分析每年的最高溫度 * 每年的最高溫度 * Created by YJ on 2017/2/7. */object case1 { def main(args: Array[String]): Unit = { val conf = new SparkConf().setMaster("local").setAppName("reduce") val sc = new SparkContext(conf) sc.setLogLevel("ERROR") //獲取數(shù)據(jù) val one = sc.textFile("hdfs://192.168.109.130:8020//user/flume/ClassicCase/case1") //過(guò)濾數(shù)據(jù) //0 0670119909 9999195005 1507004888 8888805000 01N9+00781 +999999999 9999999999 999 val yearAndTemp = one.filter(line => { //匹配0、1、4、5、9 ，有其他含義 val quality = line.substring(50, 51); var airTemperature = 0 //溫度： +零上，-零下 if(line.charAt(45)=='+'){ //00781 airTemperature = line.substring(46, 50).toInt }else{ //-00781 airTemperature = line.substring(45, 50).toInt } //airTemperature是9999為異常數(shù)據(jù)，排除，quality必須要是01459 ，數(shù)據(jù)過(guò)濾 airTemperature != 9999 && quality.matches("[01459]")}) .map{ //數(shù)據(jù)挑選，挑選年份和溫度 line =>{ val year = line.substring(15,19) var airTemperature = 0 if(line.charAt(45)=='+'){ airTemperature = line.substring(46, 50).toInt }else{ airTemperature = line.substring(45, 50).toInt } (year,airTemperature) } } //自己實(shí)現(xiàn)： PRintln("--") val groupe = yearAndTemp.groupByKey() val groupedTop5 = groupe.map(rdd => { //每個(gè)Key,取最大的一個(gè),排序 (rdd._1,rdd._2.toList.sortWith(_ > _ ).take(5)) }) groupedTop5.foreach(println) println("--") //案例實(shí)現(xiàn)：數(shù)據(jù)排序 val res = yearAndTemp.reduceByKey( (x,y)=> if(x>y) x else y ) res.collect.foreach(x=>println("year : " + x._1+", max : "+x._2)) //相同key的值相加,傳給key的值 yearAndTemp.reduceByKey(_+_).foreach(println) yearAndTemp.reduceByKey((x,y)=> x+y).foreach(println) //相同的key的值相比較，取大的值 yearAndTemp.reduceByKey( (x,y) => if(x>y) x else y ).collect.foreach(x=>println("year : " + x._1+", max : "+x._2)) sc.stop() }}上面為了過(guò)濾非法的數(shù)據(jù)，在map前先做了filter過(guò)濾。 spark執(zhí)行的任務(wù)結(jié)果： 自己過(guò)程： (1949,List(111) (1950,List(78, 22, 0, 0, -11)) 案例過(guò)程： year : 1949, max : 111 year : 1950, max : 78 推理過(guò)程： (1949,111) (1950,89) (1949,111) (1950,89) year : 1949, max : 111 year : 1950, max : 78
新聞熱點(diǎn)
疑難解答
圖片精選
