第1关：数据清洗--过滤字段长度不足的且将出生日期转：

package com.yy

 

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object edu{

    /**********Begin**********/

    // 此处可填写相关代码

    case class Person(id:String,Name:String,CtfTp:String,CtfId:String,Gender:String,Birthday:String,Address:String,Zip:String,Duty:String,Mobile:String,Tel:String,Fax:String,EMail:String,Nation:String,Taste:String,Education:String,Company:String,Family:String,Version:String,Hotel:String,Grade:String,Duration:String,City:String)

    /**********End**********/

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

        val spark = SparkSession

        .builder()

        .appName("Spark SQL")

        .master("local")

        .config("spark.some.config.option", "some-value")

        .getOrCreate()

        val rdd = spark.sparkContext.textFile("file:///root/files/part-00000-4ead9570-10e5-44dc-80ad-860cb072a9ff-c000.csv")

        /**********Begin**********/

        // 清洗脏数据（字段长度不足 23 的数据视为脏数据）

        val rdd1: RDD[String] = rdd.filter(x=>{

        val e=x.split(",",-1)

        e.length==23  })

        // 将出生日期改为 xxxx-xx-xx 格式（例如 19000101：1900-01-01，如果该属性为空不做处理，结果只取前 10 行）

        val rdd2: RDD[Person] = rdd1.map(x=>{val str=x.split(",",-1)

            if (str(5).trim != "" && str(5).length == 8) {

                str(5) = str(5).substring(0,4)+"-"+str(5).substring(4,6)+"-"+str(5).substring(6,8)

            }

            Person(str(0),str(1),str(2),str(3),str(4),str(5),str(6),str(7),str(8),str(9),str(10),str(11),str(12),str(13),str(14),str(15),str(16),str(17),str(18),str(19),str(20),str(21),str(22))

        })

        import spark.implicits._

        val df =rdd2.toDS()

        df.createOrReplaceTempView("yy")

        val out= spark.sql("select * from yy limit 10")

        // 将结果保存成 csv 格式到 file:///root/files-out 目录下

        out.write.csv("file:///root/files-out")

        /**********End**********/

        spark.stop()

      }

}

第二关：数据分析--通过入住时间和入住总时长计算用户

package com.yy

 

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object edu1{

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

    val spark = SparkSession

      .builder()

      .appName("Spark SQL")

      .master("local")

      .config("spark.some.config.option", "some-value")

      .getOrCreate()

          /**********Begin**********/

          //加载第一关处理后的数据，数据位于/root/files2目录下，文件名为part-00000-f9f4bd23-1776-4f84-9a39-f83840fa1973-c000.csv

    val df = spark.read.option("header", true).csv("file:///root/files2/part-00000-f9f4bd23-1776-4f84-9a39-f83840fa1973-c000.csv")

    //通过入住时间和入住总时长计算用户离开时间（入住时间或者入住总时长为空的不做计算）

    df.createOrReplaceTempView("yy")

    val df2: DataFrame =spark.sql("select Name,from_unixtime(unix_timestamp(Version)+Duration*3600,'yyyy-MM-dd HH:mm:ss') from yy where Version  != '' and Duration != '' limit 10")

    //将结果保存成csv格式到file:///root/files-out1目录下

    df2.write.csv("file:///root/files-out1")

       /**********End**********/ 

    spark.stop()

  }

}

第三关：数据分析--酒店被入住次数最多的3家和他们的平

package com.yy

 

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object edu2{

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

    val spark = SparkSession

      .builder()

      .appName("Spark SQL")

      .master("local")

      .config("spark.some.config.option", "some-value")

      .getOrCreate()

         /**********Begin**********/

   //加载第一关处理后的数据，数据位于/root/files3目录下，文件名为part-00000-f9f4bd23-1776-4f84-9a39-f83840fa1973-c000.csv

  val df = spark.read.option("header", true).csv("file:///root/files3/part-00000-f9f4bd23-1776-4f84-9a39-f83840fa1973-c000.csv")

    //酒店被入住次数最多的10家和他们的平均得分以及所在城市（评分为空的不做计算,注意考虑连锁酒店的情况，即同一家酒店开设在不同的城市） 

    df.createOrReplaceTempView("yy")

    val df2: DataFrame =spark.sql("select City,Hotel,avg from (select count(Hotel)as num ,Hotel,City ,round(avg(Grade),2) as avg from yy  where Grade != '' group by Hotel,City ) aa order by num desc limit 3")      

   //将结果保存成csv格式到file:///root/files-out2目录下

   df2.write.csv("file:///root/files-out2")

     /**********End**********/ 

    spark.stop()

  }

}

第四关：数据分析--每个用户每年去酒店次数及入住总时长

package com.yy

 

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object edu3{

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

    val spark = SparkSession

      .builder()

      .appName("Spark SQL")

      .master("local")

      .config("spark.some.config.option", "some-value")

      .getOrCreate()

          /**********Begin**********/

    //加载第一关处理后的数据，数据位于/root/files4目录下，文件名为part-00000-f9f4bd23-1776-4f84-9a39-f83840fa1973-c000.csv

     val df = spark.read.option("header", true).csv("file:///root/files4/part-00000-f9f4bd23-1776-4f84-9a39-f83840fa1973-c000.csv")

    //每个用户每年去酒店次数及入住总时长

    df.createOrReplaceTempView("yy")

    val df2: DataFrame =spark.sql("  select Name ,count(Id),sum(Duration),time from ( select Name ,Id,Duration,year(Version) as time  from yy where Version != '' ) a group by time,Name limit 10")  

    //将结果保存成csv格式到file:///root/files-out3目录下

    df2.write.csv("file:///root/files-out3")

       /**********End**********/

    spark.stop()

  }

}

觉得有帮助的小伙伴给个好评点赞吧转发也可以w!