【大数据技术Hadoop+Spark】Spark RDD创建、操作及词频统计、倒排索引实战（超详细附源码）

待我称王封你为后i 2023-09-30 18:03 49阅读 0赞

> **需要源码和数据集请点赞关注收藏后评论区留言私信~~~**

## 一、RDD的创建 ##

Spark可以从Hadoop支持的任何存储源中加载数据去创建RDD，包括本地文件系统和HDFS等文件系统。我们通过Spark中的SparkContext对象调用textFile()方法加载数据创建RDD。

1、从文件系统加载数据创建RDD

从运行结果反馈的信息可以看出，wordfile是一个String类型的RDD，或者以后可以简单称为RDD\[String\]，也就是说，这个RDD\[String\]里面的元素都是String类型

scala> val test=sc.textFile("file:///export/data/test.txt")
    test: org.apache.spark.rdd.RDD[String]=file:///export/data/test.txt MapPartitionsRDD[1] at textFile  at <console>:24

2、从HDFS中加载数据创建RDD

scala> val testRDD=sc.textFile("/data/test.txt")
    testRDD:org.apache.spark.rdd.RDD[String]=/data/test.txt MapPartitionsRDD[1] at textFile at  <console>:24

上面两种创建RDD的方式是完全等价的，只不过使用了不同的目录形式

Spark可以通过并行集合创建RDD。即从一个已经存在的集合、数组上，通过SparkContext对象调用parallelize()方法创建RDD。

scala> val array=Array(1,2,3,4,5)
    array: Array[Int]=Array(1,2,3,4,5)
    scala> val arrRDD=sc.parallelize(array)
    arrRDD: org.apache.spark.rdd.RDD[Int]=ParallelcollectionRDD[6] at parallelize at <console>:26

## 二、RDD的操作 ##

Spark用Scala语言实现了RDD的API，程序开发者可以通过调用API对RDD进行操作处理。RDD经过一系列的“转换”操作，每一次转换都会产生不同的RDD，以供给下一次“转换”操作使用，直到最后一个RDD经过“行动”操作才会被真正计算处理，并输出到外部数据源中，若是中间的数据结果需要复用，则可以进行缓存处理，将数据缓存到内存中

![22082083ecae4170b419c8c832b23433.png][]

### 1：转换算子 ###

RDD处理过程中的“转换”操作主要用于根据已有RDD创建新的RDD，每一次通过Transformation算子计算后都会返回一个新RDD，供给下一个转换算子使用。直到最后一个RDD经过行动操作才会被真正计算处理，并输出外部数据源中，若是中间的数据结果需要复用，则可以进行缓存处理，将数据缓存到内存中

### 2：行动算子 ###

行动算子主要是将在数据集上运行计算后的数值返回到驱动程序，从而触发真正的计算。

![13cf7c6ecec045e08e14b44b4e538236.png][]

## 三、词频统计实战 ##

在Linux本地系统的根目录下，有一个words.txt文件，文件里有多行文本，每行文本都是由2个单词构成，且单词之间都是用空格分隔。现在，通过RDD统计每个单词出现的次数（即词频），具体操作过程如下图所示。

![e233160f5d534ef381b8491645f31660.png][]

输出结果如下

![59f22963239a4a8384cd4887284234ad.png][]

## 四、倒排索引实战 ##

下面我们用Spark的RDD操作实现倒排索引功能 具体操作步骤如下

1）利用IntelliJ IDEA新建一个maven工程

2）修改pom.XML添加相关依赖包

3）在工程名处点右键，选择Open Module Settings

4）配置Scala Sdk

5) 新建文件夹scala

6) 将文件夹设置成Sources Root

7) 新建scala类

程序输出结果如下

![51bc4b348bb44315919beaef22b9cfad.png][]

代码如下

import org.apache.spark.sql.SparkSession
    
    object InvertedIndex {
      def main(args: Array[String]): Unit = {
        //获取sparkSession对象
        val spark = SparkSession.builder().appName("InvertedIndex").master("local").getOrCreate()
        //读取目录
        val data = spark.sparkContext.wholeTextFiles("D:/file")
          data.foreach(println)
    
        val r1 = data.flatMap {
             x =>
               //使用分割"/''获取文件名
               val doc=x._1.split("/").last
               //先按行切分,在按列空格进行切分
               x._2.split("\r\n").flatMap(_.split(" ").map { y => (y, doc)})}
        r1.foreach(println)
        //按单词分组
        val result=r1.groupByKey.map{case(x,y)=>(x,y.toSet.mkString(":"))}
        result.foreach(println)
      }
    }

**创作不易 觉得有帮助请点赞关注收藏~~~**

[22082083ecae4170b419c8c832b23433.png]: https://img-blog.csdnimg.cn/22082083ecae4170b419c8c832b23433.png
[13cf7c6ecec045e08e14b44b4e538236.png]: https://img-blog.csdnimg.cn/13cf7c6ecec045e08e14b44b4e538236.png
[e233160f5d534ef381b8491645f31660.png]: https://img-blog.csdnimg.cn/e233160f5d534ef381b8491645f31660.png
[59f22963239a4a8384cd4887284234ad.png]: https://img-blog.csdnimg.cn/59f22963239a4a8384cd4887284234ad.png
[51bc4b348bb44315919beaef22b9cfad.png]: https://img-blog.csdnimg.cn/51bc4b348bb44315919beaef22b9cfad.png