通過編寫Spark程序清洗酒店數(shù)據(jù)里的缺失數(shù)據(jù)、非法數(shù)據(jù)、重復(fù)數(shù)據(jù)

準(zhǔn)備工作

1. 搭建 hadoop 偽分布或 hadoop 完全分布
2. 上傳 hotal_data.csv 文件到 hadoop
3. idea 配置好 scala 環(huán)境

刪除缺失值 >= 3 的數(shù)據(jù)

1. 讀取 /hotel_data.csv
2. 刪除缺失值 >= 3 的數(shù)據(jù)， 打印剔除的數(shù)量
3. 將清洗后的數(shù)據(jù)保存為/hotelsparktask1

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object Demo01 {def main(args: Array[String]): Unit = {// System.setProperty("HADOOP_USER_NAME", "root")//解決保存文件權(quán)限不夠的問題val config: SparkConf = new SparkConf().setMaster("local[1]").setAppName("1")val sc = new SparkContext(config)val hdfsUrl ="hdfs://192.168.226.129:9000"val filePath: String = hdfsUrl+"/file3_1/hotel_data.csv"val data: RDD[Array[String]] = sc.textFile(filePath).map(_.split(",")).cache()val total: Long = data.count()val dataDrop: RDD[Array[String]] = data.filter(_.count(_.equals("NULL")) <= 3)println("刪除的數(shù)據(jù)條目有: " + (total - dataDrop.count()))dataDrop.map(_.mkString(",")).saveAsTextFile(hdfsUrl+ "/hotelsparktask1")sc.stop()}
}

刪除星級、評論數(shù)、評分中任意字段為空的數(shù)據(jù)

1. 讀取 /hotel_data.csv
2. 將字段{星級、評論數(shù)、評分}中任意字段為空的數(shù)據(jù)刪除, 打印剔除的數(shù)量
3. 保存 /hotelsparktask2

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object Demo02 {def main(args: Array[String]): Unit = {System.setProperty("HADOOP_USER_NAME", "root")val config: SparkConf = new SparkConf().setMaster("local[1]").setAppName("2")val sc = new SparkContext(config)val hdfsUrl ="hdfs://192.168.226.129:9000"val filePath: String = hdfsUrl+"/file3_1/hotel_data.csv"val data: RDD[Array[String]] = sc.textFile(filePath).map(_.split(",")).cache()val total: Long = data.count()val dataDrop: RDD[Array[String]] = data.filter {arr: Array[String] =>!(arr(6).equals("NULL") || arr(10).equals("NULL") || arr(11).equals("NULL"))}println("刪除的數(shù)據(jù)條目有: " + (total - dataDrop.count()))dataDrop.map(_.mkString(",")).saveAsTextFile(hdfsUrl+ "/hotelsparktask2")sc.stop()}
}

刪除非法數(shù)據(jù)

1. 讀取第一題的 /hotelsparktask1
2. 剔除數(shù)據(jù)集中評分和星級字段的非法數(shù)據(jù)，合法數(shù)據(jù)是評分[0，5]的實數(shù)，星級是指星級字段內(nèi)容中包含 NULL、二星、三星、四星、五星的數(shù)據(jù)
3. 剔除數(shù)據(jù)集中的重復(fù)數(shù)據(jù)
4. 分別打印 刪除含有非法評分、星級以及重復(fù)的數(shù)據(jù)條目數(shù)
5. 保存 /hotelsparktask3

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object Demo03 {def main(args: Array[String]): Unit = {System.setProperty("HADOOP_USER_NAME", "root")//解決權(quán)限問題val config: SparkConf = new SparkConf().setMaster(  "local[1]").setAppName("3")val sc = new SparkContext(config)val hdfsUrl ="hdfs://192.168.226.129:9000"val filePath: String = hdfsUrl+"/hotelsparktask1"val lines: RDD[String] = sc.textFile(filePath).cache()val data: RDD[Array[String]] = lines.map(_.split(","))val total: Long = data.count()val dataDrop: RDD[Array[String]] = data.filter {arr: Array[String] =>try {(arr(10).toDouble >= 0) && (arr(10).toDouble <= 5)} catch {case _: Exception => false}}val lab = Array("NULL", "一星", "二星", "三星", "四星", "五星")val dataDrop1: RDD[Array[String]] = data.filter { arr: Array[String] =>var flag = falsefor (elem <- lab) {if (arr(6).contains(elem)) {flag = true}}flag}val dataDrop2: RDD[String] = lines.distinctprintln("刪除的非法評分?jǐn)?shù)據(jù)條目有: " + (total - dataDrop.count()))println("刪除的非法星級數(shù)據(jù)條目有: " + (total - dataDrop1.count()))println("刪除重復(fù)數(shù)據(jù)條目有: " + (total - dataDrop2.count()))val wordsRdd: RDD[Array[String]] = lines.distinct.map(_.split(",")).filter {arr: Array[String] =>try {(arr(10).toDouble >= 0) && (arr(10).toDouble <= 5)} catch {case _: Exception => false}}.filter { arr: Array[String] =>var flag = falsefor (elem <- lab) {if (arr(6).contains(elem)) {flag = true}}flag}wordsRdd.map(_.mkString(",")).saveAsTextFile(hdfsUrl + "/hotelsparktask3")sc.stop()}
}

hotel_data.csv

下載數(shù)據(jù)：https://download.csdn.net/download/weixin_44018458/87437211