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

核心知识——论文总结

news 来源：原创 2025/7/19 23:06:04

引入

本文我们会针对论文中的核心内容进行总结，加深小伙伴对于Spark的理解。而通过Spark的论文，重点需要掌握理解如下内容：

Spark 里核心的 RDD 是一个什么概念，它是通过什么方式来优化分布式数据处理的，它的设计思路和计算机科学中的哪些观念比较相像。
Spark 在系统设计层面，是怎么针对正常情况下和异常情况下的性能进行权衡和选择的。

MapReduce的问题

在上一篇我们有提到，Spark论文的核心就是针对MapReduce明显的不足之处做了改造，在论文的引言里面提到了如下内容：

传统的集群计算框架（如MapReduce和Dryad）虽然在大规模数据分析中被广泛应用，但它们缺乏对分布式内存的有效利用，导致在需要重用中间结果的应用中效率低下。为了解决这一问题，作者提出了弹性分布式数据集（RDDs）这一新的抽象概念。RDDs具有容错性，允许用户显式地将中间结果持久化到内存中，并通过粗粒度转换来高效地实现容错，避免了数据复制和日志记录带来的高开销。基于RDDs的Spark系统在迭代应用中表现出色，比Hadoop快20倍，并且能够支持交互式数据挖掘等新应用。

我们来回忆一下MapReduce的处理流程（细节可见深入MapReduce——全流程重点梳理）：Map 函数的输出结果会输出到所在节点的本地硬盘上，Reduce 函数会从 Map 函数所在的节点里拉取它所需要的数据，然后再写入本地。接着通过一个外部排序的过程，把数据进行分组。最后将排序分完组的数据，通过Reduce 函数进行处理。

可以看到在MapReduce处理过程中，任何一个中间环节，都需要去读写硬盘。Map 函数的处理结果，并不会直接通过网络发送给 Reduce 所在的 Worker 节点，Reduce 也不会直接在内存中做数据排序。

而且这还只是单个的MapReduce任务，如论文中提到的，要通过MapReduce来跑一些机器学习任务，比如通过L-BFGS这样的算法，去进行大规模的逻辑回归的模型训练，就需要跑上百个MapReduce的任务。其中每一个MapReduce的任务都差不多，都是把所有日志读入，然后和一个梯度向量做计算，算出新的梯度。接着下一轮计算要再读入一遍原来的日志数据，再和新的梯度做计算。在这个过程中，我们原始的日志，会被重复读取上百遍。这个效率是很差的。

那很容易想到，既然读写磁盘这么浪费时间，那都通过内存处理是不是就行了？

这时候就得考虑，在这个过程中， Map或者Reduce的节点出现故障了怎么办？

因为Reduce对于前面的Map函数有依赖关系，所以任何一个Map节点故障，意味着Reduce只收到了部分数据，而且它还不知道是哪一部分。那么Reduce任务只能失败掉，然后等Map节点重新来过。而且，Reduce的失败，还会导致其他的Map节点计算的数据也要重来一遍，引起连锁反应，最终等于是整个任务重来一遍。这个成本可以说是巨大的。

而且这只是我们尝试让数据不需要落地到硬盘处理中，会遇到的一种情况，我们还可能遇到网络拥塞、内存不足以处理传输的数据等种种情况。事实上传统的MPI分布式计算系统就是这样，让一个节点直接往另外一个节点发送消息来传递数据的，但是这样的系统，容错能力很差，所以集群的规模往往也上不去。

而MapReduce针对这个问题的解决方案非常简单粗暴，那就是把整个数据处理的环节完全拆分开来，然后把一个个阶段的中间数据都落地到硬盘上。这样针对单个节点的故障，我们只需要重新运行对应节点分配的任务就好了，其他已经完成的工作不会“半途而废”。

不过这样的方式很明显不是最优解。

因为虽然在分布式场景下，我们的整个集群规模很大，节点很多，各个节点会出现故障的概率是必然的。但毕竟硬件也好、系统也好，出现故障始终还是小概率事件。正常一个集群里，年度的损坏率通常也就是 1% 左右。我们为了这 1% 的错误，却要把 99% 的数据都反复从硬盘里读出来写进去，感觉的确有点划不来。

Spark的解法

所以很自然地，我们需要有一个更有效率的容错方式。而Spark论文里面总结起来，针对容错做了以下三件事：

第一个，提供把数据缓存在内存里的能力。因为如果是计算的中间结果，我们不一定要把它写到硬盘上。如果是反复读取的输入数据，我们可以缓存在内存里而不是每个迭代重新读取一遍。
第二个，记录我们运算数据生成的“拓扑图”。也就是记录数据计算的依赖关系，一旦某个节点故障，导致这个依赖关系中的一部分节点出现故障，我们根据拓扑图重新计算这一部分数据就好了。通过这样的方式来解决容错问题，而不是每一次都把数据写入到硬盘。
第三个，通过检查点来在特定环节把数据写入到硬盘。当我们的拓扑图层数很深，或者数据要反复进行很多次的迭代计算。前面通过“拓扑图”进行重新计算的容错方式会变得非常低效，那么我们可以在一部分中间环节，把数据写入到硬盘上。这样一种折衷的方式，既避免了每次都去从硬盘读写数据，也避免了一旦某一个环节出现故障，“容错”方案只能完全从头再来的尴尬出现。

以前面提到的大规模逻辑回归的机器模型训练举例，原本需要进行 100轮迭代。我们最好的解决方案，既不是每轮迭代都需要重新读写数据，那样太浪费硬盘的 I/O 了。也不是把几小时的计算过程都放在内存里，那样万一计算梯度结果的数据在第 99 轮丢失了，我们就要从头开始。我们完全可以把读取的日志数据缓存在内存里，然后把每 10 轮计算完的梯度数据写入到硬盘上。这样一旦出现故障，我们只需要重新读取一次日志数据，并最多计算 10轮迭代的过程就好了。

论文 3.2.1 部分使用 Spark 实现分布式逻辑回归的代码如下：

val points = spark.textFile(...).map(parsePoint).persist()
var w = // random initial vector
for (i <- 1 to ITERATIONS) {val gradient = points.map{ p =>p.x * (1/(1+exp(-p.y*(w dot p.x)))-1)*p.y}.reduce((a,b) => a+b)w -= gradient
}

其中输入数据会通过 .persistent 缓存在内存中，而不需要每个迭代都重新从硬盘读取。其实这个思路，就是 Spark 整个系统设计的出发点。根据这个朴素的思路，Spark 定了一个新的概念RDD，全称是 Resilient Distributed Dataset，中文叫做弹性分布式数据集。整个系统设计的其实就是 “弹性”+“分布式”+“数据集” 这三点的组合。分布式和数据集这两个关键字，很好理解，本质就是采用了数据分区的方式，来确保数据是分布式的。

RDD

而 RDD 最核心的设计关键点，就在这个弹性上。论文里的 2.1 部分，给出了 RDD 明确的定义：

Formally, an RDD is a read-only, partitioned collection of records. RDDs can only be created through deterministic operations on either (1) data in stable storage or (2) other RDDs. We call these operations transformations to differentiate them from other operations on RDDs. Examples of transformations include map, filter, and join.
RDDs do not need to be materialized at all times. Instead, an RDD has enough information about how it was derived from other datasets (its lineage) to compute its partitions from data in stable storage. This is a powerful property: in essence, a program cannot reference an RDD that it cannot reconstruct after a failure.

Finally, users can control two other aspects of RDDs: persistence and partitioning. Users can indicate which RDDs they will reuse and choose a storage strategy for them (e.g., in-memory storage). They can also ask that an RDD’s elements be partitioned across machines based on a key in each record. This is useful for placement optimizations, such as ensuring that two datasets that will be joined together are hash-partitioned in the same way.

Although individual RDDs are immutable, it is possible to implement mutable state by having multiple RDDs to represent multiple versions of a dataset. We made RDDs immutable to make it easier to describe lineage graphs, but it would have been equivalent to have our abstraction be versioned datasets and track versions in lineage graphs.

翻译：

正式来说，弹性分布式数据集（RDD）是一个只读的、分区的记录集合。RDD只能通过两种方式创建：（1）对稳定存储中的数据进行确定性操作，或（2）对其他RDD进行操作。我们称这些操作为转换，以区分它们与其他RDD操作。转换的例子包括映射、过滤和连接。

RDD不需要始终被物化。相反，RDD拥有足够的信息来记录它是如何从其他数据集（其谱系）中派生的，从而可以从稳定存储中的数据计算出它的分区。这是一个强大的特性：本质上，程序无法引用在故障后无法重建的RDD。

最后，用户可以控制RDD的另外两个方面：持久化和分区。用户可以指示哪些RDD将被重用，并选择它们的存储策略（例如，内存存储）。他们还可以要求根据每个记录中的键将RDD的元素分区到不同的机器上。这对于放置优化非常有用，例如确保将要连接的两个数据集以相同的方式进行哈希分区。

虽然单个RDD是不可变的，但可以通过使用多个RDD来表示数据集的不同版本来实现可变状态。我们将RDD设计为不可变的，以便更容易描述谱系图，但也可以将我们的抽象设计为版本化数据集，并在谱系图中跟踪版本。

其中核心就是，RDD 是只读的、已分区的记录集合，RDD 只能通过明确的操作，以及通过两种数据创建：稳定存储系统中的数据；其他 RDD。这个明确的操作，是指 map、filter 和 join 这样的操作，以和其他的操作区分开来。

按照这个定义，我们可以看到这个是对于数据的一个抽象。我们的任何一个数据集，进行一次转换就是一个新的 RDD，但是这个 RDD 并不需要实际输出到硬盘上。实际上这个数据都不会作为一个完整的数据集缓存在内存中，而只是一个 RDD 的“抽象概念”。只有当我们对某一个 RDD 实际调用 persistent 函数的时候，这个 RDD 才会实际作为一个完整的数据集，缓存在内存中。

一旦被缓存到内存里，这个 RDD 就能够再次被下游的其他数据转换反复使用。一方面，这个数据不需要写入到硬盘，所以我们减少了一次数据写。另一方面，下游的其他转化也不需要再从硬盘读数据，于是，我们就节省了大量的硬盘 I/O 的开销。

我们可以对照着论文 2.2.1 中的示例代码，来看这样一个过程：

lines = spark.textFile("hdfs://...")
errors = lines.filter(_.startsWith("ERROR"))
errors.persist()// Count errors mentioning MySQL:
errors.filter(_.contains("MySQL")).count()// Return the time fields of errors mentioning
// HDFS as an array (assuming time is field
// number 3 in a tab-separated format):
errors.filter(_.contains("HDFS"))
.map(_.split(’\t’)(3))
.collect()

我们从 HDFS 上，读入原始数据，根据关键词 ERROR 进行了一次过滤，然后把它 persistent 下来。而接下来分别有两个分析任务，会用到这个缓存在内存里的 ERROR 数据，一个是找出所有带有 MySQL 关键词的错误日志，然后进行统计行数；另一个则是找到所有带有HDFS 关键字的日志，然后按照 Tab 分割并收集第 3 列的数据。

如上，是论文中的整个流程图，一开始从 HDFS 里读入的 line 数据，因为没有 persistent，所以不会缓存在内存中。而 errors 会缓存在内存里面，供后面两个任务作为输入使用。

errors 我们不需要写入到硬盘里，而后面分析 MYSQL 和 HDFS 关键字错误的两个任务，也不需要从硬盘读数据，数据都是直接在内存中读写，所以性能大大加快了。

从 RDD 的这个逻辑上，其实我们可以看到计算机工程上的其他系统中的影子。

第一个是惰性求值（Lazy-Evaluation），我们的一层层数据转化，只要没有调用persistent，都可以先不做计算，而只是记录这个计算过程中的函数。而当 persistent 一旦被调用，那么我们就需要把实际的数据结果计算出来，并存储到内存里，再供后面的数据转换程序调用。
第二个是数据库里的视图功能。为了查询方便，对于复杂的多表关联，很多时候我们会预先建好一张数据库的逻辑视图。那么我们在查询逻辑视图的时候，其实还是通过一个多表关联 SQL 去查询原始表的，这个就好像我们并没有调用 persistent，把数据实际持久化下来。当然，我们也可以把对应视图的查询结果，直接写入一张中间表，这样实际上就相当于把计算的结果持久化下来了，后续查询的 SQL 就会查询这个中间表。如果视图里的数据会被后续的 SQL 反复多次查询，并且对应的原始数据集也和 RDD 一样是不可变的话，一样会大大提升系统整体的效率。

宽依赖关系和检查点

虽然通过调用 persistent 来把数据缓存到内存里，减少了大量的硬盘读写，但是我们仍然会面临节点失效，导致 RDD 需要重新计算的情况。

所以 Spark 对这部分流程做了进一步的优化，这个优化说起来其实也不复杂。那就是如果一个节点失效了，导致的数据重新计算，需要影响的节点太多，那么我们就把计算结果输出到硬盘上。而如果影响的节点少，那么我们就只是单独重新计算被响应到的那些节点就好了。

所以在 Spark 里，会对整个数据计算的拓扑图在分布式系统下的依赖关系做一个分类。如果一个 RDD 的一个分区，只会影响到下游的一个节点，那么我们就称这样的上下游依赖关系为窄依赖。而如果一个 RDD 的一个分区，会影响到下游的多个节点，那么我们就称这样的上下游关系为宽依赖。

对于窄依赖，即使重算一遍，也只是影响一条线上的少数几个节点，所以对应的中间数据结果，并不需要输出到硬盘上。
而对于宽依赖，一旦上游的一个节点失效了，需要重新计算。那么它对应的多个下游节点，都需要重新从这个节点拉取数据并重新计算，需要占用更多的网络带宽和计算资源。换句话说，在宽依赖下，一个上游节点的失效，会以几倍的影响在下游得到放大。所以，在宽依赖的场景下，上游会像 MapReduce 里的 Map 一样，把输出结果序列化到硬盘上，以减少故障后的恢复成本。

同样的，对于有多轮迭代，或者是整个拓扑图很长的数据处理任务，Spark 在persistent 的时候，支持你添加一个 REPLICATE 参数，把当前的计算结果作为一个检查点存储下来。一旦添加了这个参数，数据就不只是存储在内存中，而是会序列化到硬盘里。这样，同样可以减少你在出现故障时候的重新计算的时间。

可以看到，无论是 persistent、宽依赖下的数据会被持久化存储，还是允许用户去自己通过检查点存储中间步骤的计算结果，其本质都是为了在性能和容错间做一个平衡。如果不做任何持久化存储，那么平时系统会跑得很快，但是一旦某个节点出错就要从头再来。而如果都做持久化存储，那么节点出错的时候，计算可以恢复得很快，但是没有问题的时候会有很多浪费。Spark的操作就是在尽可能平衡的同时，给用户更多选择的，以适应不同的业务场景需求。

总结

通过本文的梳理总结，可以看到Spark里的RDD的设计思想并不复杂。和MapReduce一样，RDD的设计思路也是来自于函数式编程。相对于过程式地把每一个数据转换（Transformation）的结果存储下来，RDD相当于记录了输入数据，以及对应的计算输入数据的函数。

而这个方式，和把一步步的计算结果存储下来的效果一样，都可以解决容错问题。当某一个RDD的某一个分区因为故障丢失的时候，我们可以通过输入数据和对应的函数，快速计算出当前RDD的实际内容。而这个输入数据+对应函数的组合，就是RDD中的DAG Lineage图。

RDD和其他分布式系统最大的差异，就在代表弹性的R这个关键字上。

这个弹性体现在两个方面：

第一个是数据存储上。数据不再是存放在硬盘上的，而是可以缓存在内存中。只有当内存不足的时候，才会把它们换出到硬盘上。同时，数据的持久化，也支持硬盘、序列化后的内存存储，以及反序列化后Java对象的内存存储三种形式。虽然需要占用更多的内存，但是计算速度会更快。
第二个是选择把什么数据输出到硬盘上。 Spark会根据数据计算的DAG Lineage，来判断某一个RDD对于前置数据是宽依赖，还是窄依赖的。如果是宽依赖的，意味着一个节点的故障，可能会导致大量的数据要进行重新计算，乃至数据网络传输的需求。那么，它就会把数据计算的中间结果存储到硬盘上。

同时，Spark也支持用户定义检查点，能够基于自己的业务场景，把一些关键节点的数据通过检查点的方式，持久化到硬盘上，避免出现特定节点的故障，导致大量数据需要重新计算的问题。

总的说起来，Spark的RDD 支持了MapReduce可以支持的所有运算方式。并且还通过尽可能利用内存，使得需要多个MapReduce的组合或者迭代的任务的执行速度大大加快了。从论文里展示的分布式逻辑回归处理效果来看，Spark的性能会比使用原始的MapReduce一轮轮迭代快上20倍。这也是为什么，Spark一出现在市场上，就很快替代了大量的MapReduce的分析工作，并在迭代式的机器学习算法中成为了主流。

引入

MapReduce的问题

Spark的解法

RDD

宽依赖关系和检查点

总结

相关文章：