企业官网建设流程全解析

1. 从一次数据清洗任务说起：为什么需要Spark

几年前我接手过一个日志清洗的活儿，原始数据是每天几百GB的文本文件，散落在几十台机器的磁盘上。当时第一反应是用脚本逐行读、逐行处理，结果跑了一晚上才处理完一天的数据，第二天数据量翻倍，直接崩了。后来换成多进程，勉强能跑，但机器一多，进程间通信和容错就成了噩梦——某台机器挂了，整个任务得从头再来。

这个场景其实很典型：数据量大到单机扛不住，但又不是大到需要动用专门的大数据集群。Spark就是在这个夹缝里杀出来的。它不像Hadoop MapReduce那样每个步骤都要落盘，而是尽量把中间结果放在内存里流转，速度能快上一个数量级。更关键的是，它把分布式计算的复杂度封装得相当好，你写代码的感觉跟写单机程序差不多，但背后自动帮你切分数据、调度任务、处理失败重试。

这篇文章面向的是刚接触Spark的开发者，或者之前只用过Pandas、NumPy这类单机工具，现在需要处理更大规模数据的人。我会从架构层面把Spark拆开讲清楚，重点落在RDD这个最核心的抽象上。不堆概念，而是讲清楚每个设计决策背后的“为什么”，以及在实际写代码时这些设计会怎么影响你的选择。

2. Spark架构的骨架：Driver、Executor和集群管理器

2.1 一个Spark应用到底由哪些角色组成

先抛开那些花哨的术语，一个Spark应用运行时，本质上就是一个Driver进程加上一组Executor进程。Driver是你写main函数的地方，它负责把用户代码转换成任务，然后调度到各个Executor上执行。Executor是真正干活的，它们分布在集群的不同节点上，每个Executor内部有若干个CPU核心和一块内存区域，用来跑任务和缓存数据。

你可以把Driver想象成工头，Executor想象成工人。工头手里拿着图纸（你的代码），把活儿拆成一个个小任务，然后分配给工人。工人干完活把结果汇报给工头，工头再决定下一步干什么。这个比喻虽然粗糙，但能帮你理解一个关键点：Driver是整个应用的单点，它挂了整个应用就挂了。所以生产环境里Driver通常要配置高可用，或者至少保证它所在的节点稳定。

集群管理器是另一个角色，它负责在物理资源层面分配Executor。Spark支持三种模式：Standalone、YARN和Mesos（现在用得少了）。Standalone是Spark自带的简易调度器，适合测试和小规模部署；YARN是Hadoop生态里的资源管理器，企业里用得最多；Mesos曾经很火，但现在社区活跃度下降明显。选哪个主要看你现有的基础设施——如果已经有YARN集群，直接用YARN最省事。

2.2 为什么Spark要把内存管理做得这么细

Spark和MapReduce最大的区别在于中间结果不落盘。MapReduce的每一步Map和Reduce之间都要把数据写到HDFS，下一步再从HDFS读回来。这个设计保证了容错性——任何一步失败，只要输入数据还在，就能重算。但代价是大量的磁盘I/O，速度上不去。

Spark的做法是尽量把中间数据留在内存里。但内存是有限的，不可能把所有数据都塞进去。所以Spark把Executor的内存划分成几个区域：Execution内存用于Shuffle、Join、Sort等操作时的临时数据；Storage内存用于缓存RDD；User Memory留给用户代码里的数据结构；还有一小块Reserved Memory给Spark内部对象用。

这个划分不是静态的。在统一内存管理模式下，Execution和Storage可以互相借用。如果Execution不够了，它会去抢Storage的空闲部分；反过来Storage不够时也能借Execution的。这个机制的好处是灵活，坏处是调优时你得盯着两个区域的使用情况，不然容易出现一边闲置一边OOM的尴尬局面。

实际调优时，我一般先把spark.memory.fraction设成0.6到0.7，然后观察Storage内存的占用。如果缓存RDD经常被驱逐，就适当调大；如果Shuffle频繁 spill 到磁盘，就得考虑增加Executor总内存或者减少每个Executor的核心数。

2.3 集群管理器选型：Standalone、YARN还是Kubernetes

Standalone模式最简单，启动一个Master和若干Worker就能跑。但它没有资源隔离，多个应用之间会互相抢资源，而且不支持动态资源分配。适合开发测试或者小团队内部用。

YARN模式是企业主流。它的优势在于资源调度成熟，能和Hadoop生态无缝集成，支持队列、优先级、动态资源分配。缺点是配置相对复杂，而且YARN的调度延迟比Standalone高一些。如果你的集群已经有YARN在跑其他任务，用YARN是最自然的选择。

Kubernetes模式是近几年的新趋势。它把Spark应用打包成容器，利用K8s的调度和弹性伸缩能力。好处是资源隔离彻底，部署标准化，适合云原生环境。缺点是网络和存储的配置需要额外考虑，而且Driver的Pod如果挂了，恢复起来比YARN麻烦一些。

选型时我的建议是：看团队现有的运维能力。如果团队对YARN很熟，就别折腾K8s；如果已经在全面容器化，那K8s模式更顺理成章。Standalone只建议在学习和测试阶段用。

3. RDD：Spark最核心的抽象到底抽象了什么

3.1 弹性分布式数据集的三层含义

RDD全称是Resilient Distributed Dataset，中文叫弹性分布式数据集。这个名字里的每个词都有实际含义，不是随便起的。

Resilient（弹性）指的是容错能力。RDD记录了它是怎么从其他RDD转换来的（这叫血统Lineage），如果某个分区的数据丢了，Spark可以根据血统重新计算那个分区，而不需要把整个数据集重算。这个设计比检查点机制轻量得多，也是Spark能高效容错的关键。

Distributed（分布式）指的是数据是切分存储的。一个RDD在逻辑上是一个整体，但物理上被分成多个Partition，分散在集群的不同节点上。每个Partition是一段连续的数据，Spark的任务调度就是以Partition为单位的。

Dataset（数据集）指的是它本质上是一个数据集合，你可以对它做各种转换和动作。但注意，RDD是不可变的——你不能修改一个已有的RDD，只能通过转换生成新的RDD。这个不可变性是Spark能轻松实现容错和并行计算的基础。

3.2 转换与动作：惰性求值的实际影响

RDD的操作分两类：Transformation（转换）和Action（动作）。转换是惰性的，调用map、filter、flatMap这些方法时，Spark并不会立即执行，而是记录下这个操作，生成一个新的RDD。只有当你调用count、collect、saveAsTextFile这些动作时，Spark才会真正触发计算。

这个惰性求值的设计有两个好处。第一，它允许Spark做整体优化。比如你连续调用了map和filter，Spark可以把它们合并成一个阶段，避免中间数据的产生。第二，它让流水线执行成为可能。在一个Stage内部，数据可以一条一条地流过整个转换链，不需要等所有数据都处理完再进入下一步。

但惰性求值也有坑。最常见的问题是：你在循环里反复对一个RDD调用动作。比如下面这种写法：

rdd = sc.textFile("hdfs://path/to/data") for i in range(10): count = rdd.filter(lambda x: x.contains(str(i))).count() print(count)

这段代码会触发10次完整的计算，每次都要从HDFS重新读数据、重新执行filter。正确的做法是先做一次转换，把结果缓存起来：

rdd = sc.textFile("hdfs://path/to/data").cache() for i in range(10): count = rdd.filter(lambda x: x.contains(str(i))).count() print(count)

cache()会把RDD的数据缓存在内存里，后续的filter直接基于缓存数据计算，速度会快很多。但注意，cache()本身也是惰性的，它不会立即缓存，而是在第一次动作触发时才缓存。

3.3 宽依赖与窄依赖：Shuffle为什么昂贵

RDD之间的依赖关系分两种：窄依赖和宽依赖。窄依赖是指父RDD的每个Partition最多被一个子RDD的Partition使用，比如map、filter、union。宽依赖是指父RDD的每个Partition可能被多个子RDD的Partition使用，比如groupByKey、reduceByKey、join。

窄依赖的好处是可以在同一个Stage内流水线执行，不需要Shuffle。宽依赖则必须等父RDD的所有Partition都计算完，然后根据key重新分发数据，这个过程就是Shuffle。Shuffle涉及磁盘I/O、网络传输和序列化，是Spark作业中最昂贵的操作。

理解这一点对写高效Spark代码至关重要。能避免Shuffle就避免，不能避免就尽量减少Shuffle的数据量。比如用reduceByKey代替groupByKey，因为reduceByKey会在Shuffle之前先在每个Partition内做局部聚合，大大减少网络传输的数据量。再比如做join时，如果有一张表很小，可以用广播变量把它分发到每个Executor，避免Shuffle。

我见过一个典型的性能问题：两个大表做join，其中一个表其实只有几MB，但代码里直接用了join，导致全量Shuffle。改成broadcast之后，作业时间从40分钟降到3分钟。这个教训值得每个Spark开发者记住。

4. 从代码到执行：一个Spark作业的完整生命周期

4.1 编写代码时到底发生了什么

当你写下一段Spark代码并提交时，Driver首先会解析你的代码，构建出一个DAG（有向无环图）。这个DAG的节点是RDD，边是转换操作。然后Driver会把这个DAG提交给DAG调度器，DAG调度器根据宽依赖的位置把DAG切分成多个Stage。每个Stage内部是一系列窄依赖的转换，可以流水线执行。

Stage分两种：ShuffleMapStage和ResultStage。ShuffleMapStage的输出是Shuffle所需的数据，ResultStage的输出是最终结果。一个作业可能包含多个ShuffleMapStage和一个ResultStage，取决于有多少次Shuffle。

Stage切分完之后，每个Stage会被进一步拆分成TaskSet，每个Task对应一个Partition的处理。TaskSet交给Task调度器，Task调度器再把Task分配到各个Executor上执行。Executor收到Task后，启动一个线程去执行，执行过程中如果需要读取Shuffle数据，就从本地或远程的Shuffle服务拉取。

4.2 任务调度中的本地性级别

Task调度器在分配Task时，会尽量考虑数据本地性。本地性从高到低分几个级别：PROCESS_LOCAL（数据在同一个Executor进程的内存里）、NODE_LOCAL（数据在同一个节点的磁盘上）、RACK_LOCAL（数据在同一个机架）、ANY（数据在任意地方）。

理想情况下，Task应该被分配到数据所在的节点，这样就不需要网络传输。但现实中资源可能不够，或者数据分布不均匀，调度器需要在本地性和资源利用率之间做权衡。Spark的策略是：先等一会儿，看能不能等到本地资源；如果等不到，就降级到下一个本地性级别。

这个等待时间由spark.locality.wait控制，默认是3秒。如果你的作业对延迟很敏感，可以调小这个值；如果网络带宽是瓶颈，可以调大，让调度器多等一会儿本地资源。

4.3 容错机制：血统、检查点和推测执行

Spark的容错主要靠血统。每个RDD都记录了它是从哪些父RDD经过什么转换得来的。如果某个Partition丢了，Spark就沿着血统往上找，重新计算那个Partition。这个过程对用户是透明的，不需要你写任何额外的代码。

但血统有个问题：如果依赖链很长，重算的代价会很大。这时候可以用检查点。检查点是把RDD的数据写到可靠的存储（比如HDFS）上，切断血统链。后续如果数据丢了，直接从检查点恢复，不需要从头重算。检查点适合那些计算代价很高、依赖链很长的RDD。

推测执行是另一个容错机制。如果某个Task执行得特别慢（可能是因为所在节点负载高或者硬件有问题），Spark会在另一个节点上启动一个相同的Task，谁先完成就用谁的结果。这个机制能有效缓解“长尾任务”拖慢整个作业的问题。但推测执行也有代价——它会消耗额外的资源，而且在某些情况下（比如Task有副作用）可能导致重复计算。所以默认是关闭的，需要手动开启。

5. 写RDD代码时最容易踩的几个坑

5.1 闭包序列化问题

Spark的转换操作里经常需要引用外部变量，比如：

class MyClass: def __init__(self): self.factor = 10 def process(self, rdd): return rdd.map(lambda x: x * self.factor)

这段代码在Driver上定义，但map里的lambda会在Executor上执行。Spark需要把lambda以及它引用的self序列化后发送到Executor。如果MyClass不可序列化，就会报NotSerializableException。

解决办法有几个：把需要的变量提取成局部变量，避免引用整个对象；或者让类实现Serializable接口；或者用broadcast把共享数据分发出去。我一般推荐第一种，因为最简单也最不容易出问题。

5.2 在转换里创建大对象

另一个常见问题是在转换函数里创建大对象。比如：

rdd.map(lambda x: SomeHeavyObject().process(x))

每次调用lambda都会创建一个新的SomeHeavyObject，如果这个对象初始化很慢，或者占用内存很大，就会严重影响性能。正确的做法是在Driver上创建一次，然后通过广播变量分发：

heavy_obj = SomeHeavyObject() broadcast_obj = sc.broadcast(heavy_obj) rdd.map(lambda x: broadcast_obj.value.process(x))

广播变量会把对象缓存到每个Executor上，所有Task共享同一个实例，避免了重复创建的开销。

5.3 滥用collect和count

collect()会把所有数据拉到Driver上，如果数据集很大，Driver直接OOM。count()虽然只返回一个数字，但它会触发全量计算。这两个动作在调试时很方便，但在生产代码里要慎用。

如果只是想看几条数据，用take(10)；如果想统计数量，考虑用reduce或者aggregate做分布式聚合；如果确实需要全量数据，先确认Driver的内存够不够，或者用saveAsTextFile写到分布式存储上。

5.4 分区数设置不合理

分区数太少，每个分区数据量太大，容易OOM，而且并行度不够；分区数太多，Task数量过多，调度开销大，而且每个Task处理的数据太少，效率低。一般来说，每个CPU核心处理2到4个分区比较合适。如果集群有100个核心，分区数设在200到400之间。

调整分区数用repartition或coalesce。repartition会做全量Shuffle，分区数可以增加也可以减少；coalesce默认不做Shuffle，只能减少分区数。如果是从HDFS读数据，分区数默认等于文件的分块数，通常不需要调整。如果是经过多次转换后分区数变得不合理，再用repartition或coalesce调整。

6. 从RDD到DataFrame：什么时候该换工具

RDD是Spark最基础的抽象，灵活但底层。你需要自己管理序列化、自己优化执行计划。DataFrame和Dataset是更高层的抽象，它们带有Schema信息，Spark的Catalyst优化器可以自动做很多优化，比如谓词下推、列裁剪、常量折叠。

如果你的数据是结构化的（有明确的列和类型），优先用DataFrame。它的性能通常比手写RDD好，代码也更简洁。RDD适合那些非结构化数据、需要精细控制执行过程、或者DataFrame表达不了的场景。

我个人的经验是：新项目直接用DataFrame API，老项目里已有的RDD代码不用急着改。DataFrame的优化器确实能省很多事，但RDD的灵活性在某些场景下仍然不可替代。理解RDD的原理，能帮你更好地理解DataFrame底层在做什么，遇到性能问题时也知道从哪个方向排查。

最后分享一个排查Spark作业性能问题的通用思路：先看Spark UI里的Stage和Task分布，找出耗时最长的Stage；然后看那个Stage里有没有Shuffle，Shuffle的数据量有多大；接着看Task的执行时间分布，有没有长尾Task；最后看内存使用情况，有没有频繁的spill或者GC。按这个顺序走一遍，大部分性能问题都能定位到。