Spark编程题

现有100W+条数据存储在hdfs中的userinfo文件夹中的多个文件中,数据格式如下：

张三|男|23|未婚|北京|海淀

李四|女|25|已婚|河北|石家庄

求：

1.数据中所有人的平均年龄

2.数据中所有男性未婚的人数和女性未婚人数

3.数据中20-30已婚数量前3的省份

答案：

package spark08

import org.apache.spark.rdd.RDD

import org.apache.spark.util.LongAccumulator

import org.apache.spark.{SparkConf, SparkContext}

/**

*张三|男|23|未婚|北京|海淀

*李四|女|25|已婚|河北|石家庄

  *

*统计:

* 1.数据中所有人的平均年龄

* 2.数据中所有男性未婚的人数和女性未婚人数

* 3.数据中20-30已婚数量前3的省份

* 4.未婚比例(未婚人数/该城市总人数)最高的前3个城市

  */

object UserInfo {

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

    val conf: SparkConf = new SparkConf()

      .setAppName(this.getClass.getSimpleName)

      .setMaster("local[*]")

    val sc = new SparkContext(conf)

//读取原始文件

    val strFile: RDD[String] = sc.textFile("D:\\data\\data\\userinfo")

    val srcRdd: RDD[(String, String, Int, String, String, String)] = strFile.map(t => {

      val strings: Array[String] = t.split("\\|")

      val name: String = strings(0)

      val gender = strings(1)

      val age = strings(2).toInt

      val isMarry: String = strings(3)

      val province = strings(4)

      val city = strings(5)

      (name, gender, age, isMarry, province, city)

    })

    srcRdd.cache()

//1.数据中所有人的平均年龄 李四|女|25|已婚|河北|石家庄

val ageAccumulator: LongAccumulator = sc.longAccumulator //使用累加器统计总人数

    val ageCount: Int = srcRdd.map(t => {

      ageAccumulator.add(1)

      t._3

    }).reduce(_ + _)

    val ageNumber = ageAccumulator.value

    val avgAge = ageCount.toLong/(ageNumber*1.0)

println(s"所有人的平均年龄为${avgAge}")

//2.数据中所有男性未婚的人数和女性未婚人数

    val genderAndMarryRDD: RDD[(String, Iterable[(String, String)])] = srcRdd.map(t => {

(t._1, t._3) //性别,婚否

}).filter(_._2.equals("未婚")).groupBy(_._1) //按性别分组

    val res2RDD: RDD[(String, Int)] = genderAndMarryRDD.mapValues(t=>t.size)

    res2RDD.collect().foreach(println)

//数据中20-30已婚数量前3的省份 李四|女|25|已婚|河北|石家庄

    val res3: Array[(Int, String)] = srcRdd.filter(t => {

t._3 >= 20 && t._3 <= 30 && t._4.equals("已婚")

})//删选出满足20-30已婚的数据，按省份分组，求v的size即是20-30已婚数量

      .groupBy(_._5).mapValues(_.size)

//k,v互换取前3

      .map(t => (t._2, t._1)).top(3)

    res3.foreach(println)

//(城市,(未婚人数,已婚人数))

//未婚比例(未婚人数/该城市总人数)最高的前3个城市 李四|女|25|已婚|河北|石家庄