Spark-核心编程（一）模拟分布式计算以及RDD基础原理

快来打我* 2022-09-02 11:55 182阅读 0赞

# Spark核心编程 #

Spark 计算框架为了能够进行高并发和高吞吐的数据处理，封装了三大数据结构，用于处理不同的应用场景。三大数据结构分别是：

RDD : 弹性分布式数据集

累加器：分布式共享只写变量

广播变量：分布式共享只读变量  
接下来我们一起看看这三大数据结构是如何在数据处理中使用的。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]

## 模拟分布式计算 ##

task

class Task extends Serializable { 
    
        val datas = List(1, 2, 3, 5)
    
        val logic: Int => Int = _ *2
    }

subTask

class SubTask extends Serializable { 
    
        var datas: List[Int] = _
    
        var logic: Int => Int = _
    
        // 计算
        def compute(): List[Int] = { 
            val ints = datas.map(logic)
            ints
        }
    }

Driver

object Driver { 
    
        def main(args: Array[String]): Unit = { 
            // 启动客户端，发送数据
            val client1 = new Socket("localhost", 9999)
            val client2 = new Socket("localhost", 8888)
    
            val task = new Task
    
            val out1 = client1.getOutputStream
            val objOut1 = new ObjectOutputStream(out1)
            val subTask1 = new SubTask
            subTask1.datas = task.datas.take(2)
            subTask1.logic = task.logic
            objOut1.writeObject(subTask1)
            objOut1.flush()
            objOut1.close()
            println("client1 data send over")
            client1.close()
    
            val out2 = client2.getOutputStream
            val objOut2 = new ObjectOutputStream(out2)
            val subTask2 = new SubTask
            subTask2.datas = task.datas.takeRight(2)
            subTask2.logic = task.logic
            objOut2.writeObject(subTask2)
            objOut2.flush()
            objOut2.close()
            println("client2 data send over")
            client2.close()
        }
    }

Executor

object Executor { 
        def main(args: Array[String]): Unit = { 
            // 启动服务器，接受数据
            val server = new ServerSocket(9999)
            println("server 9999 is running")
    
            // 等待客户端的链接
            val client = server.accept()
    
            val in = client.getInputStream
            val objIn = new ObjectInputStream(in)
    
            val task = objIn.readObject().asInstanceOf[SubTask]
            val result = task.compute()
    
            println(s"ger client send data and compute result is $result")
    
            objIn.close()
            client.close()
            server.close()
        }
    }

Executor2

object Executor2 { 
        def main(args: Array[String]): Unit = { 
            // 启动服务器，接受数据
            val server = new ServerSocket(8888)
            println("server 8888 is running")
    
            // 等待客户端的链接
            val client = server.accept()
    
            val in = client.getInputStream
            val objIn = new ObjectInputStream(in)
    
            val task = objIn.readObject().asInstanceOf[SubTask]
            val result = task.compute()
    
            println(s"ger client send data and compute result is $result") // ger client send data and compute result is List(2, 4, 6, 10)
    
            objIn.close()
            client.close()
            server.close()
        }
    }

## RDD ##

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是 Spark 中**最基本的数据处理模型**。代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

1、弹性：

存储的弹性: 内存与磁盘的自动切换;

容错的弹性: 数据丢失可以自动恢复;

计算的弹性: 计算出错重试机制;

分片的弹性: 可根据需要重新分片。

2、分布式: 数据存储在大数据集群不同节点上

3、数据集: RDD 封装了计算逻辑,并不保存数据

4、数据抽象: RDD 是一个抽象类,需要子类具体实现

5、不可变: RDD 封装了计算逻辑,是不可以改变的,想要改变,只能产生新的 RDD,在新的 RDD 里面封装计算逻辑

6、可分区、并行计算

WordCount的RDD执行流程如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]

### 核心属性 ###

![在这里插入图片描述][45e07a24061545108a62160fc4a6ca44.png]

分区列表：RDD 数据结构中存在分区列表,用于执行任务时并行计算,是实现分布式计算的重要属性

![在这里插入图片描述][4e1c62cf206c4d43bb0d62911b299a17.png]

分区计算函数：Spark 在计算时,是使用分区函数对每一个分区进行计算，每个分区里面的计算逻辑是一样的

![在这里插入图片描述][2981146c0c0b440f834c0d8820c8e74f.png]

RDD 之间的依赖关系：RDD 是计算模型的封装,当需求中需要将多个计算模型进行组合时,就需要将多个 RDD 建立依赖关系  
![在这里插入图片描述][9d9ee000276e4968a719f8b9f6d92b85.png]

分区器(可选)：当数据为 KV 类型数据时,可以通过设定分区器自定义数据的分区，也就是通过分区规则进行分区处理  
![在这里插入图片描述][adc3de5ba4294f749bc69e87b3448672.png]

首选位置(可选)：计算数据时,可以根据计算节点的状态选择不同的节点位置进行计算，用来判断计算发送到哪个节点效率最优；移动数据不如移动计算  
![在这里插入图片描述][237eab1649da427e99aa3e6f3f60fcf6.png]

### 执行原理 ###

从计算的角度来讲,数据处理过程中需要计算资源(内存 & CPU)和计算模型(逻辑)。执行时,需要将计算资源和计算模型进行协调和整合。

Spark 框架在执行时,先申请资源,然后将应用程序的数据处理逻辑分解成一个一个的计算任务。然后将任务发到已经分配资源的计算节点上, 按照指定的计算模型进行数据计算。最后得到计算结果。

RDD 是 Spark 框架中用于数据处理的核心模型,接下来我们看看,在 Yarn 环境中, RDD的工作原理:

1、启动 Yarn 集群环境

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 2]

2、Spark 通过申请资源创建调度节点和计算节点

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 3]

3、Spark 框架根据需求将计算逻辑根据分区划分成不同的任务

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 4]

4、调度节点将任务根据计算节点状态发送到对应的计算节点进行计算

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 5]

从以上流程可以看出 RDD 在整个流程中主要用于将逻辑进行封装,并生成 Task 发送给Executor 节点执行计算,接下来我们就一起看看 Spark 框架中 RDD 是具体是如何进行数据处理的。

### 基础编程 ###

#### RDD 创建 ####

在 Spark 中创建 RDD 的创建方式可以分为四种:

##### 从集合(内存)中创建 RDD #####

从集合中创建 RDD,Spark 主要提供了两个方法:parallelize 和 makeRDD，从底层代码实现来讲,makeRDD 方法其实就是 parallelize 方法

object TestRDDMemory { 
    
        def main(args: Array[String]): Unit = { 
            // 1、准备环境
            // 这里的local[*]表示根据计算机的所有核创建对应个数的线程来模拟真实计算场景，不设置默认是1
            val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
            val sc = new SparkContext(sparkConf)
    
            // 2、创建RDD
            // 从内存中创建RDD，将内存中集合的数据作为处理的数据源
            val seq = Seq[Int](1, 2, 3, 4)
            // parallelize：并行
            // val rdd: RDD[Int] = sc.parallelize(seq)
            // makeRDD方法在底层实现其实就是调用了rdd对象的parallelize方法。
            val rdd: RDD[Int] = sc.makeRDD(seq)
    
            // 不调用collect的话也能执行，但是顺序会乱掉
            rdd.collect().foreach(println)
    
            // 3、关闭环境
            sc.stop()
    
        }
    }

##### 从外部存储(文件)创建 RDD #####

由外部存储系统的数据集创建 RDD 包括:本地的文件系统,所有 Hadoop 支持的数据集,比如 HDFS、HBase 等。

// textFile：以行为单位读取数据，读取的数据都是字符串
    object TestRDDFile { 
    
        def main(args: Array[String]): Unit = { 
            // 1、准备环境
            // 这里的local[*]表示根据计算机的所有核创建对应个数的线程来模拟真实计算场景，不设置默认是1
            val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
            val sc = new SparkContext(sparkConf)
    
            // 2、创建RDD
            // 从文件中创建RDD，将文件中的数据作为处理的数据源
            // path路径默认以当前项目的根路径作为基准，可以写绝对路径，也可以写相对路径
            // val rdd: RDD[String] = sc.textFile("datas/scala/test")
    
            // path可以是文件名也可以是目录名，如果是文件夹的话，要注意该文件夹下面不能再有文件夹，否则会报错
            // val rdd: RDD[String] = sc.textFile("datas/scala")
    
            // path也可以使用通配符
            // val rdd: RDD[String] = sc.textFile("datas/scala/test*")
    
            // path也可以是分布式存储系统路径：HDFS
            val rdd: RDD[String] = sc.textFile("hdfs://hadoop113:8020/sanguo/wei")
    
    
            // 不调用collect的话也能执行，但是顺序会乱掉
            rdd.collect().foreach(println)
    
            // 3、关闭环境
            sc.stop()
    
        }
    }
    
    // wholeTextFiles：以文件为单位读取数据，
    // 读取的结果表示为元组，第一个元素表示文件路径，第二个元素表示文件内容
    object TestRDDFile1 { 
    
        def main(args: Array[String]): Unit = { 
            // 1、准备环境
            // 这里的local[*]表示根据计算机的所有核创建对应个数的线程来模拟真实计算场景，不设置默认是1
            val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
            val sc = new SparkContext(sparkConf)
    
            // 2、创建RDD
            // 从文件中创建RDD，将文件中的数据作为处理的数据源
            val rdd = sc.wholeTextFiles("datas/scala")
    
    
            // 不调用collect的话也能执行，但是顺序会乱掉
            rdd.collect().foreach(println)
            // (file:/home/lxj/workspace/Spark/datas/scala/out,hello hahahahaha)
            // (file:/home/lxj/workspace/Spark/datas/scala/test,hello caocao
            // hello sanguo
            // hello donghan)
            // (file:/home/lxj/workspace/Spark/datas/scala/test1,hello caocao
            // hello sanguo
            // hello donghan)
    
            // 3、关闭环境
            sc.stop()
    
        }
    }

##### 从其他 RDD 创建 #####

主要是通过一个 RDD 运算完后,再产生新的 RDD。一般是由框架自己使用

##### 直接创建 RDD(new) #####

使用 new 的方式直接构造 RDD,一般由 Spark 框架自身使用。

#### RDD 并行度与分区 ####

默认情况下,Spark 可以将一个作业切分多个任务后,发送给 Executor 节点并行计算,而能够并行计算的任务数量我们称之为并行度。这个数量可以在构建 RDD 时指定。这里的并行执行的任务数量,并不是指的切分任务的数量。

读取内存数据时,数据可以按照并行度的设定进行数据的分区操作

object TestRDDMemoryPar { 
    
        def main(args: Array[String]): Unit = { 
            // 1、准备环境
            // 这里的local[*]表示根据计算机的所有核创建对应个数的线程来模拟真实计算场景，不设置默认是1
            val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    
            // 配置分区的数量
            // sparkConf.set("spark.default.parallelism", "5")
    
            val sc = new SparkContext(sparkConf)
    
            // 2、创建RDD
            // RDD的并行度 && 分区
            // makeRDD方法可以传递第二个参数，这个参数表示分区的数量
            // 第二个参数可以不传递，不传的话会使用默认值，defaultParallelism
            // spark在默认情况下，从配置对象中获取配置参数：spark.default.parallelism
            // 如果获取不到，那么使用totalCores属性，这个属性取值为当前运行环境的最大可用核数
            // 上面配置的是*即当前服务器的核数，如果是配置了具体值，则取其具体的值
            // 【1, 2】【3, 4】
            val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
            // 【1】【2】【3, 4】
            // val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 3)
            // 【1】【2, 3】【4, 5】
            // val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5), 3)
    
    
            // 将处理的数据保存成分区文件
            rdd.saveAsTextFile("output")
    
            // 3、关闭环境
            sc.stop()
        }
    }

数据分区规则的Spark 核心源码如下

def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = { 
        (0 until numSlices).iterator.map {  i =>
            val start = ((i * length) / numSlices).toInt
            val end = (((i + 1) * length) / numSlices).toInt
            (start, end)
        }
    }
    
    val array = seq.toArray // To prevent O(n^2) operations for List etc
    positions(array.length, numSlices).map {  case (start, end) =>
        array.slice(start, end).toSeq
    }.toSeq
    // slice(from: Int, until: Int)

读取文件数据时,数据是按照 Hadoop 文件读取的规则进行切片分区,而切片规则和数据读取的规则有些差异

object TestRDDFilePar { 
    
        def main(args: Array[String]): Unit = { 
            // 1、准备环境
            // 这里的local[*]表示根据计算机的所有核创建对应个数的线程来模拟真实计算场景，不设置默认是1
            val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
            val sc = new SparkContext(sparkConf)
    
            // 2、创建RDD
            // textFile可以讲文件作为数据处理的数据源，默认也可以设定分区
            // minPartitions：最小分区数量
            // math.min(defaultParallelism, 2)
            // 如果不想使用默认的分区数量，可以通过第二个参数指定分区数
            // Spark读取文件，底层是Hadoop的读取方式，也就是每次读取一行
            // 分区数量的计算方式：
            // totalSize = 7（1和2后面携带了回车换行符（占用2字节），因此1+2+1+2+1=7）
            // goalSize = 7 / 2 = 3(byte) 每个分区3个字节
            // 7 / 3 = 2余1（这里也会涉及到分区1.1倍的概念） + 1 = 3
    
            // 数据分区的分配
            // 1、数据以行为单位进行读取
            // spark读取文件，采用的是hadoop的方式读取，所以一行一行读取，和字节数没有关系
            // 2、数据读取时以偏移量为单位，已经被读过的行，将不会重复读取
            // 3、数据分区的偏移量范围的计算
            // 0 => [0, 3] => 12
            // 1 => [3, 6] => 3
            // 2 => [6, 7] =>
            // 【1, 2】【3】【】
            val rdd: RDD[String] = sc.textFile("datas/scala/test2", 2)
            // test2的数据为
            // 1 => 偏移量为012
            // 2 => 偏移量为345
            // 3 => 偏移量为6
    
            // 分区计算
            // 14 / 2 = 7（byte）
            // 14 / 7 = 2 个分区
            // 数据分区的偏移量范围的计算
            // [0, 7] => 01234567
            // [7, 14] => 890
            // val rdd: RDD[String] = sc.textFile("datas/scala/test3", 2)
            // test3的数据为
            // 1234567 => 偏移量为012345678
            // 89 => 偏移量为9101112
            // 0 => 偏移量为13
    
            // 如果数据源为多个文件，那么计算分区时以文件为单位进行分区
    
            rdd.saveAsTextFile("output")
    
            // 3、关闭环境
            sc.stop()
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]: /images/20220829/fa20f9f8b5ca48d88d3ddcd795ed3ec9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]: /images/20220829/64aadcc0ecf240bda8965d2f928d12ea.png
[45e07a24061545108a62160fc4a6ca44.png]: /images/20220829/27f76f61a9174842acb11d697522e602.png
[4e1c62cf206c4d43bb0d62911b299a17.png]: /images/20220829/bd811c5f274a49f38c1c0d7c54c01899.png
[2981146c0c0b440f834c0d8820c8e74f.png]: /images/20220829/a3dbc38e89b24c2299acdd8374a4901c.png
[9d9ee000276e4968a719f8b9f6d92b85.png]: /images/20220829/fdee95266d9c4723800d1d4c40d9814d.png
[adc3de5ba4294f749bc69e87b3448672.png]: /images/20220829/2180cedbb842468f9799788698fc2f98.png
[237eab1649da427e99aa3e6f3f60fcf6.png]: /images/20220829/cf3c013324414d79a9b1b84100ecfa51.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 2]: /images/20220829/d1ce4d4965014d51a082c6573bc2e60a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 3]: /images/20220829/4006a3ca179241d182e9638436f1b602.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 4]: /images/20220829/567b90b479ed403ea146a78a9f8c3804.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 5]: /images/20220829/a65328ea3a0c4d818286a2147b81ac5b.png