案例数据清洗
(一)需求说明
【项目的需求】
假设你有一个包含用户信息的文本文件,每行格式为 姓名,年龄,性别,需要清洗掉年龄为空或者非数字的行。
以下是 10 条符合上述示例中数据格式(姓名,年龄,性别)的测试数据,包含了一些可能需要清洗掉的无效数据,你可以将其保存为一个文本文件,用于测试上面的数据清洗程序。
张三,25,男
李四,,女
王五,30,男
赵六,a,女
孙七,35,男
周八,40,女
吴九,abc,男
郑十,45,女
王十,50,男
李二,55,女
这里面:“李四” 的年龄为空,“赵六” 和 “吴九” 的年龄不是有效的数字,在执行数据清洗程序时,这些行应该会被过滤掉。
(二)思路分析
读入文件
对每一行数据进行分析
字段拆分,拆分出年龄这个字段
判断
如果它不是数字或者缺失,则忽略这条数据
否则保存
(三)难点突破
读入txt文件。 val lines = sc.textFile(inputFile)
对拆分出一行中的年龄。val fields = line.split(",") fields(0)
过滤算子中,函数返回为false,就会被过滤掉,函数返回为true,就会被保留下来。
使用正则表达式。/\d/
(四)功能实现
创建新的maven项目。
创建input文件夹,在input下新建记事本文件,内容就是前面的实例数据。
在src下创建新的scala文件,开始写功能代码。
// 1.读入文件
// 2.开始过滤,对每一行数据进行分析
// (1)字段拆分,拆分出年龄这个字段
// (2)判断
// ①如果它不是数字或者缺失,则忽略这条数据
// ②否则保存
// 3.保存过滤之后的文件
(五)参考代码
import org.apache.spark.{SparkConf, SparkContext}
object DataCleaning {
def main(args: Array[String]): Unit = {
// 创建 SparkConf 对象
val conf = new SparkConf().setAppName("DataCleaning").setMaster("local[*]")
// 创建 SparkContext 对象
val sc = new SparkContext(conf)
// 读取文本文件,创建 RDD
val inputFile = "input/file.txt"
val lines = sc.textFile(inputFile)
// 数据清洗操作
val cleanedLines = lines.filter(line => { // 使用filter算子
val fields = line.split(",")
if (fields.length == 3) {
val age = fields(1).trim
age.matches("\\d+")
} else {
false
}
})
// 输出清洗后的数据
cleanedLines.collect().foreach(println)
// 停止 SparkContext
sc.stop()
}
}
【核心代码讲解】
把每行文本按逗号分隔成字段,查看字段数量是否为 3。检查年龄字段是否为有效的数字。
(六)拓展
这里介绍一个拓展点:如何把清洗之后的数据保存到一个文件中。
可以使用coalesce(1)这个方法可以让结果全部保存在一个文件中。
val singlePartitionRDD = cleanedLines.coalesce(1)
// 保存清洗后的数据到文件
val outputPath = "path/to/your/output/file.txt"
singlePartitionRDD.saveAsTextFile(outputPath)
// 停止 SparkContext
sc.stop()