企业Spark案例--酒店数据分析实战提交

第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!