spark的练习题（6）

左手的ㄟ右手 2022-04-08 13:24 128阅读 0赞

1.排序：  
定义三个文件对文件内容进行排序（数字）

package com.hyxy.spark01
    
    import org.apache.spark.{SparkConf, SparkContext}
    
    object listarray {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("sort").setMaster("local")
        val sc = new SparkContext(conf)
        val rdd = sc.textFile("D://123//sort.txt")
          rdd.sortBy(x =>x,true)
          .repartition(1)
          .foreach(println)
    
      }
    }
    //val data = sc.textFile(dataFile,3)
    //var index = 0
    //val result = data.filter(_.trim().length>0).map(n=>(n.trim.toInt,""))
    //  .partitionBy(new HashPartitioner(1))
    //  .sortByKey()
    //  .map(t=>{index += 1;(index,t._1)})
    //result.saveAsTextFile("file:///D:/测试数据/排序/result")
    //}

2.题目：给定一组键值对("spark",2),("hadoop",6),("hadoop",4),("spark",6)，键值对的key  
表示图书名称，value表示某天图书销量，请计算每个键对应的平均值，也就是计算每种图书的每天平均销量。

package com.hyxy.spark01
    
    import org.apache.spark.{SparkConf, SparkContext}
    
    object pingjunzhi {
      def main(args: Array[String]): Unit = {
    //    给定一组键值对("spark",2),("hadoop",6),("hadoop",4),("spark",6)，键值对的key
    //    表示图书名称，value表示某天图书销量，请计算每个键对应的平均值，也就是计算每种图书的每天平均销量。
        val conf = new SparkConf().setMaster("local[2]").setAppName("平均销量")
        val sc = new SparkContext(conf)
        val rdd = sc.parallelize(Array(("spark",2),("hadoop",6),("hadoop",4),("spark",6),("spark",2)))//parallelize是创建rdd的
        rdd.mapValues(x => (x,1))//("spark",(2,1))
          .reduceByKey((x,y)=> (x._1+y._1,x._2+y._2))//("spark",(8,2))
          .mapValues(x => (x._1 / x._2)).foreach(println)//("spark",4)
    
    
      }
    }

3.二次排序  
题目：要求先按账户排序，在按金额排序  
hadoop@apache 200  
hive@apache 550  
yarn@apache 580  
hive@apache 159  
hadoop@apache 300  
hive@apache 258  
hadoop@apache 150  
yarn@apache 560  
yarn@apache 260  
结果：(hadoop@apache,\[150,200,300\]),(hive@apache,\[159,258,550\]),....

package com.hyxy.spark01
    
    import org.apache.spark.{SparkConf, SparkContext}
    
    object twoarray {
      def main(args: Array[String]): Unit = {
    //    要求先按账户排序，在按金额排序
        val conf = new SparkConf().setMaster("local[2]").setAppName("二次排序")
        val sc = new SparkContext(conf)
        val rdd = sc.textFile("D:\\123\\twoarray.txt")
        .map(x => x.replaceAll("\\s+"," ").split(" "))
          .map(x => (x(0),x(1)))
          .repartition(1).groupByKey()
          .mapValues(x => x.toList.sortBy(x=>x))
          .sortByKey()
          .foreach(println)
    
      }
    }

4.TopN  
任务描述： orderid(订单ID),userid(用户ID),payment(支付金额),productid(商品ID)  
file1.txt  
1,1768,50,155  
2,1218,600,211  
3,2239,788,242  
4,3101,28,599  
5,4899,290,129  
6,3110,54,1201  
7,4436,259,877  
8,2369,7890,27  
file2.txt  
100,4287,226,233  
101,6562,489,124  
102,1124,33,17  
103,3267,159,179  
104,4569,57,125  
105,1438,37,116  
求Top N个payment支付金额值

package com.hyxy.spark01
    
    import org.apache.spark.{SparkConf, SparkContext}
    
    object TopN {
      //任务描述： orderid(订单ID),userid(用户ID),payment(支付金额),productid(商品ID)
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("TopN").setMaster("local")
        val sc = new SparkContext(conf)
        //sc.setLogLevel("EROOR")
        val lines = sc.textFile("D://123//TopN//*.txt")
        var num = 0
        val result = lines.filter(line => (line.trim().length > 0) && (line.split(",").length == 4))
          .map(_.split(",")(2))
          .map(x => (x.trim.toInt, ""))
          .sortByKey(false).map(_._1).take(5) //降序
          .foreach(x => {
          num = num + 1
          println(num + "\t" + x)
        })
      }
    }

5.单表关联  
题目：求孙子和祖父母的关系列表  
数据： child parent  
Tom Lucy  
Tom Jack  
Jone Lucy  
Jone Jack  
Lucy Mary  
Lucy Ben  
Jack Alice  
Jack Jesse  
Terry Alice  
Terry Jesse  
Philip Terry  
Philip Alma  
Mark Terry  
Mark Alma

package com.hyxy.spark01
    
    import org.apache.spark.{SparkConf, SparkContext}
    
    object SingleTable {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("单表关联").setMaster("local")
        val sc = new SparkContext(conf)
        val child_parent_rdd = sc.textFile("D://123//SingleTable.txt").filter(x=>{if (x.contains("child")) false else true})
          .map(x=>{val str=x.replaceAll("\\s+"," ").split(" ");(str(0),str(1))})
        val parent_child_rdd = child_parent_rdd.map(x => (x._2,x._1))
        val child_grand_rdd = child_parent_rdd.join(parent_child_rdd);//（父母，（祖父母，孙子））
        val grandchild_grandparent_rdd = child_grand_rdd.map(x=>(x._2._2,x._2._1)).repartition(1).foreach(println)
    
      }
    }

6.降水量案例  
讲解  
计算俄罗斯100多年的降水总量，并列出降水量最多的十年；  
数据说明：  
【20674 1936 1 1 0 -28.0 0 -24.9 0 -20.4 0 0.0 2 0 OOOO】  
0.气象站编码  
1.年  
2.月  
3.日  
4.空气温度质量标记  
5.每日最低温度  
6.每日最低温度标记：0表示正常，1表示是存疑，9表示异常或无观测值  
7.每日平均温度  
8.每日平均温度标记：0表示正常，1表示是存疑，9表示异常或无观测值  
9.每日最高温度  
10.每日最高温度标记：0表示正常，1表示是存疑，9表示异常或无观测值  
11.每日降水量  
12.每日降水量标记：0表示降水量超过0.1mm，1表示数天的统计量，2表示无观察值，2表示降水小于0.1mm  
13.每日降水量标记：0表示正常，1表示是存疑，9表示异常或无观测值  
14.数据标记：4位，AAAA表示使用新版数据规范；oooo表示数据比较值不变；RRRR表示数据比较值可变  
思路：  
step1：加载数据，进行数据清洗(过滤不符合条件的数据，包括：数据完整度、准确度)；  
step2: 获取年份和降水量，并将对年份分组，计算总降水量；  
step3：按降水量进行排序(降序操作)  
step4：获取降水量最多的前十年  
step5：将最终结果输出到HDFS中

package com.hyxy.spark01
    
    import org.apache.spark.{SparkConf, SparkContext}
    
    object jyl {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local").setAppName("jyl")
        val sc = new SparkContext(conf)
        val rdd =sc.textFile("D:\\123\\ussr")
        rdd.map(x => x.trim().replaceAll("   "," ")
          .replaceAll("  "," ").split(" "))
          .filter(_.length == 15)
          .filter(_(13)!="9")
          .filter(_(11)!="999.9")
          .map(x => (x(1),x(11).toDouble))
          .repartition(1)
          .reduceByKey(_+_)
          .map(x => (x._2,x._1))
          .sortByKey(false)
          .map(x => (x._2,x._1))
          .zipWithIndex()
          .filter(x => x._2<10)
          .map(x => (x._1._1,x._1._2))
          .saveAsTextFile("D:\\123\\JYL")
    
      }
    }

Apache *Spark 是*专为大规模数据处理而设计的快速通用的计算引擎