当前位置：首页 > news >正文

SpringBoot项目——使用Spark对爬虫爬取下的数据进行清洗

news 2025/7/17 1:29:15

随着互联网信息呈爆炸式增长，爬虫技术被广泛用于从海量网页中抓取有价值的数据。然而，爬取到的数据往往存在格式不规范、重复、噪声等诸多问题，需要高效的数据清洗流程来保障数据质量，Spark 在其中发挥了关键作用。

什么是Spark

Spark 是当今大数据领域最活跃、最热门、最高效的大数据通用计算平台之一。

Spark 是为大规模数据处理而设计的分布式计算框架，旨在处理海量数据的存储和分析任务。它可以在集群环境中运行，将计算任务分布到多个节点上，利用集群的并行处理能力来加速数据处理过程。提供了基础的弹性分布式数据集（RDD）抽象，是 Spark 的核心部分，可进行通用的分布式数据处理操作。

Spark的优点

快:与Hadoop的MapReduce相比，Spark基于内存的运算要快100倍以上；而基于磁盘的运算也要快10倍以上。Spark实现了高效的DAG执行引擎，可以通过基于内存来高效地处理数据流。
易用:Spark支持Java、Python和Scala的API，还支持超过80种高级算法，使用户可以快速构建不同应用。而且Spark支持交互式的Python和Scala的Shell，这意味着可以非常方便的在这些Shell中使用Spark集群来验证解决问题的方法，而不是像以前一样，需要打包、上传集群、验证等。这对于原型开发非常重要。
通用性：Spark提供了统一的解决方案。Spark可以用于批处理、交互式查询（通用Spark SQL）、实时流处理（通过Spark Streaming）、机器学习（通过Spark MLlib）和图计算（通过Spark GraphX）。这些不同类型的处理都可以在同一应用中无缝使用。Spark统一的解决方案非常具有吸引力，毕竟任何公司都想用统一的平台处理问题，减少开发和维护的人力成本和部署平台的物理成本。当然还有，作为统一的解决方案，Spark并没有以牺牲性能为代价。相反，在性能方面Spark具有巨大优势。
可融合性：Spark非常方便的与其他开源产品进行融合。比如，Spark可以使用Hadoop的YARN和Apache Mesos作为它的资源管理和调度器，并且可以处理所有Hadoop支持的数据，包括HDFS、HBase和Cassanda等。这对于已部署Hadoop集群的用户特别重要，因为不需要做任何数据迁移就可以使用Spark强大的处理能力。Spark也可以不依赖第三方的资源管理器和调度器，它实现了Standalone作为其内置资源管理器和调度框架，这样进一步降低了Spark的使用门槛，使得所有人可以非常容易地部署和使用Spark。此外Spark还提供了在EC2上部署Standalone的Spark集群的工具。

Spark的使用

Spark大至使用流程

要先将数据进行存放在一个txt文本文件当中，
使用Spark进行读取txt中的文本数据，进行数据处理
将清洗后的数据转存到原来的txt文件当中
想要存放到数据库当中，则将txt文本文件中的数据再次读出出来存放进去即可

Spark的maven依赖

Spark想要在Springboot项目中使用要引入相应的maven依赖

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.4.1</version>
</dependency>
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.12</artifactId><version>3.4.1</version>
</dependency>

爬取出数据的存储

本项目进行爬取的是求职网站，会将求职岗位的信息放入到一个List集合当中，通过遍历这个List集合，将数据存放到txt文本中

    /*** @param jobs 爬取下来的数据集合* @param filePath 你要存放数据的地址*/public static void writeDataToTxt(List<Job> jobs, String filePath) {//进行检查文件是否存在 若不存在则进行创建一个路径为filePath的文件File file = new File(filePath);if (!file.exists()) {try {file.createNewFile();} catch (IOException e) {throw new RuntimeException(e);}}try (BufferedWriter writer = new BufferedWriter(new FileWriter(filePath))) {// 将数据写入文件for(Job job : jobs) {//这里因为直接进行放入到txt文本文件难以进行格式处理 我们使用手动进行拼接操作String j = job.getJobName()+","+job.getCompany()+","+job.getSalary()+","+job.getExperience()+","+job.getEducational()+","+job.getCity()+","+job.getCompanyScale()+","+job.getCompanyStatus();writer.write(j);// 写入换行符writer.newLine();}} catch (IOException e) {// 处理可能出现的异常System.err.println("写入文件时出现错误: " + e.getMessage());}}

Spark的数据清洗

本项目中的Spark进行数据清洗是通过JavaRDD<E>

JavaRDD 是 Spark 为 Java 开发者提供的弹性分布式数据集（Resilient Distributed Dataset，RDD）接口。从本质上讲，它是一个分布式的集合，其中的数据被划分成多个分区（Partitions），这些分区可以分布在集群中的不同节点上进行并行处理。这种分布式的特性使得 JavaRDD 能够高效地处理大规模的数据。

JavaRDD 是 Spark Core 的重要组成部分，是构建其他高级功能（如 Spark SQL、MLlib 和 Spark Streaming）的基础。在 Spark SQL 中，数据通常被封装成 DataFrame 或 Dataset，但这些高级数据结构的底层实现往往也依赖于 RDD 的基本概念和操作。

JavaRDD 可以存储各种类型的数据，包括 Java 基本数据类型（如int、double、boolean等）和自定义的 Java 对象。

使用JavaRDD 的filter方法和map方法进行对数据的处理

filter 方法是用来对流中的元素进行筛选的，它的返回值应该是一个 boolean 类型的表达式，用来判断该元素是否应该保留在流中
map 方法是一种转换操作，它对 JavaRDD 中的每个元素应用一个函数，并返回一个新的 JavaRDD

    /*** @param path 要进行清洗的txt文件路径*/public static void DataClening(String path){SparkSession spark = SparkSession.builder()//将 SparkContext、SQLContext 和 HiveContext 等功能集成在一起.appName("Java Spark")//为 Spark 应用程序设置一个名称 名称会显示在 Spark 集群的监控界面上，有助于识别和管理应用程序.master("local[*]")//表示在本地运行 Spark.getOrCreate();//尝试获取现有的 SparkSession 实例，如果不存在则创建一个新的。这样可以确保在同一个 JVM 中不会创建多个 SparkSession，避免资源浪费JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());//读取数据源到RDD中JavaRDD<String> rdd = sc.textFile(path);//对RDD中的数据进行清洗处理JavaRDD<String> cleanedData = cleanData(rdd);// 打印清洗后的数据（可以根据需要保存到文件或进行其他操作） 此处不要省略 不然会报错cleanedData.collect().forEach(System.out::println);//进行检查文件是否存在File file = new File(path);if (file.exists()) {file.delete();}try (BufferedWriter writer = new BufferedWriter(new FileWriter(path))) {// 将数据写入文件for(String job : cleanedData.collect()) {writer.write(job);// 写入换行符writer.newLine();}} catch (IOException e) {// 处理可能出现的异常System.err.println("清洗数据后写入文件时出现错误: " + e.getMessage());}// 停止Spark上下文sc.stop();}/*** 对数据进行清洗操作* @param rawData 需要进行清洗的JavaRDD对象* @return 返回进行清洗的JavaRDD对象*/private static JavaRDD<String> cleanData(JavaRDD<String> rawData) {return rawData//filter 方法是用来对流中的元素进行筛选的，它的返回值应该是一个 boolean 类型的表达式，用来判断该元素是否应该保留在流中// 移除空行.filter(line -> !line.trim().isEmpty())// 例如，移除含有特定字符（如null）的行//.filter(line -> !line.contains("null"))// map 是一种转换操作，它对 JavaRDD 中的每个元素应用一个函数，并返回一个新的 JavaRDD// 可能需要根据分隔符（例如逗号）拆分字段并重新格式化.map(line -> {String[] fields = line.replace("null", "暂无数据信息").split(",");//进行数据检查 确保这工作名称等8项信息完整 不完整直接进行设置为空if (fields.length == 8) {//将fields字符串数组中的信息进行拼接起来 这里仅做简单拼接为清洗后的格式，实际清洗可以根据需求更复杂return String.join(",", fields);} else {return ""; //将不满足8项信息的数据进行设置为空处理}})// 去除清洗后的空行或不需要的数据.filter(line -> !line.trim().isEmpty())//去重.distinct();}

清洗后数据存放

此处会将txt文件中的数据进行取出，放入到一个List集合当中，想要放入到数据库中，只需要进行遍历这个集合即可

    /*** * @param path 将指定位置的txt文件中的信息 放入到List集合中 便于插入到数据库中* @return 返回文本中的·数据信息*/public static List<Job> getJobbyTxtFile(String path){List<Job> jobs = new ArrayList<>();try (BufferedReader reader = new BufferedReader(new FileReader(path))) {String line;while ((line = reader.readLine())!= null) {jobs.add(toJob(line));}} catch (IOException e) {e.printStackTrace();}return jobs;}public static Job toJob(String job){String[] jobsContext = job.split(",");Job j = new Job();j.setJobName(jobsContext[0]);j.setCompany(jobsContext[1]);j.setSalary(jobsContext[2]);j.setExperience(jobsContext[3]);j.setEducational(jobsContext[4]);j.setCity(jobsContext[5]);j.setCompanyScale(jobsContext[6]);j.setCompanyStatus(jobsContext[7]);return j;}

查看全文

http://www.lryc.cn/news/518271.html