企业官网建设流程全解析

1. 为什么RDD才是PySpark里最该先啃透的那块硬骨头

很多人上手PySpark的第一反应是直接冲DataFrame，觉得RDD是“老古董”，API又底层又啰嗦，不如DataFrame写SQL来得痛快。我一开始也这么想，直到有一次处理一批日志数据，格式极其不规整，每行里嵌套着多层不固定的分隔符，用DataFrame读进来全是null，schema怎么定义都不对。折腾了大半天，最后用RDD的map加split几行代码就搞定了。那次之后我才真正意识到，RDD不是过时的东西，它是Spark整个计算模型的根基，DataFrame和Spark SQL底层跑的还是RDD那一套。你理解了RDD，后面学DataFrame的惰性求值、分区并行、宽窄依赖，全都能一通百通。

这篇内容主要围绕PySpark中RDD的数据处理展开，适合已经跑通过PySpark基本环境、想深入理解RDD操作体系的同学。我会从RDD的核心数据模型讲起，把Transformation和Action的区别掰开揉碎，再逐一拆解常见的转换算子和行动算子，配上可直接运行的代码示例。最后会重点聊分区调优和几个我实际踩过的坑，这些是你在官方文档里不太容易看到的。

关键词方面，围绕RDD、Transformation、Action、分区、惰性求值、PySpark这些核心概念展开，不堆砌术语，尽量用实际场景把每个知识点串起来。

2. RDD到底是个什么东西：从数据模型到执行逻辑

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

RDD全称Resilient Distributed Dataset，中文叫弹性分布式数据集。这个名字里每个词都不是随便取的，拆开来看正好对应它的三个核心特性。

Resilient（弹性）指的是容错能力。RDD之间会记录血缘关系（Lineage），也就是每个RDD是怎么从父RDD计算出来的。如果某个分区的数据丢了，Spark不需要从头重新跑整个任务，只需要根据血缘关系重新计算那一个分区就行。这跟传统数据库靠冗余备份做容错的路子完全不同，Spark靠的是“记住怎么算出来的”，而不是“存多份”。

Distributed（分布式）指的是数据物理上分散在集群的多个节点上。一个RDD在逻辑上是一个整体，但实际存储时被切分成多个Partition（分区），每个分区分布在不同节点。你写代码时操作的是一个RDD对象，但底层并行执行时，每个分区是独立处理的。

Dataset（数据集）说明它本质上是一个数据集合，你可以把它理解成一个分布式的列表。跟普通Python列表最大的区别在于，这个列表里的元素你没法直接索引访问，只能通过转换操作来批量处理。

用一个生活化的类比：把RDD想象成一列火车上的货物清单。清单本身是一张纸（逻辑视图），但货物实际分散在几十个车厢里（分区）。你没法直接伸手去拿第37个箱子，但你可以给每个车厢的搬运工下一道指令（Transformation），让他们各自处理自己车厢里的货，最后统一汇总（Action）。

2.2 RDD的不可变性与血缘关系

RDD有一个非常关键的特性：不可变性。你没法修改一个已有的RDD，每次转换操作都会生成一个新的RDD。这听起来好像很浪费空间，但实际上Spark并不会真的把每个中间RDD都物化存储，它只记录转换的步骤，等到遇到Action时才真正触发计算。

举个例子：

rdd1 = sc.parallelize([1, 2, 3, 4, 5]) rdd2 = rdd1.map(lambda x: x * 2) rdd3 = rdd2.filter(lambda x: x > 4)

这段代码执行完，rdd1、rdd2、rdd3三个对象都存在，但实际的数据计算一次都没发生。Spark只是在rdd3上记录了一条血缘链：rdd3是从rdd2过滤来的，rdd2是从rdd1映射来的。直到你调用比如rdd3.collect()，Spark才会沿着这条链回溯，把计算真正跑一遍。

这种设计的好处是容错和优化。容错前面说了，优化方面，Spark可以在真正执行前对整条血缘链做分析，比如把连续的map合并成一个阶段，减少数据 shuffle。

注意：不可变性意味着你不能对RDD做原地修改。如果你需要一个“更新后”的结果，必须用转换操作生成新RDD。这在写代码时要注意变量命名，别把新旧RDD搞混了。

2.3 分区：RDD并行计算的基本单位

分区是RDD并行度的体现。一个RDD有多少个分区，就意味着最多能有多少个任务同时处理它。分区数量直接决定了你的任务跑得快不快。

创建RDD时，分区数可以手动指定：

# 不指定分区数，Spark根据集群配置自动决定 rdd = sc.parallelize([1, 2, 3, 4, 5]) # 手动指定4个分区 rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=4) # 从HDFS文件创建，分区数默认等于文件的分块数 rdd = sc.textFile("hdfs://path/to/file.txt") # 手动指定分区数 rdd = sc.textFile("hdfs://path/to/file.txt", minPartitions=10)

分区数不是越多越好，也不是越少越好。分区太少，并行度不够，集群资源浪费；分区太多，每个任务调度开销累积起来也很可观。经验值是每个分区处理128MB到256MB的数据比较合适，这个数字跟HDFS块大小对齐，能减少数据跨节点传输。

你可以用rdd.getNumPartitions()查看分区数，用rdd.glom().collect()查看每个分区里的实际数据分布。后者在调试数据倾斜时特别有用。

3. Transformation与Action：惰性求值到底省了什么

3.1 两类算子的本质区别

PySpark的RDD操作分成两大类：Transformation（转换）和Action（行动）。这个分类不是随便分的，它直接对应Spark的惰性求值机制。

Transformation是“定义操作”，它接收一个RDD，返回一个新的RDD，但不会立即执行。常见的map、filter、flatMap、distinct、groupByKey、reduceByKey都属于这一类。

Action是“触发执行”，它接收一个RDD，返回一个具体的结果（不是RDD），比如一个数字、一个列表、或者把数据写到外部存储。常见的collect、count、first、take、reduce、saveAsTextFile属于这一类。

为什么这么设计？因为惰性求值让Spark有机会在真正执行前做全局优化。比如你连续写了三个map，Spark可以把它们合并成一个阶段，避免中间结果的物化和传输。如果每个Transformation都立即执行，这种优化就不可能了。

用一个类比：Transformation就像在餐厅点菜，你告诉服务员要什么，但厨房还没开始做。Action就像你说“可以上菜了”，厨房才开始真正动手。点菜过程中你随时可以改主意，加菜减菜都行，一旦下了“上菜”的指令，厨房就按最终菜单一次性做出来。

3.2 惰性求值的实际影响

惰性求值有一个很实际的影响：报错时机。如果你写了一个有问题的Transformation，比如lambda里引用了不存在的变量，代码在定义时不会报错，只有遇到Action时才会抛出异常。

rdd = sc.parallelize([1, 2, 3]) # 这行不会报错，因为map是Transformation rdd2 = rdd.map(lambda x: x / 0) # 这行才会报错，因为collect是Action # rdd2.collect() # ZeroDivisionError

这个特性在调试时要注意。你写完一串Transformation，最好紧接着加一个Action来验证，别等到最后才发现中间某一步有问题。

另一个影响是重复计算。因为RDD不物化中间结果，如果你对同一个RDD调用多次Action，每次都会从头重新计算整条血缘链。

rdd = sc.parallelize([1, 2, 3, 4, 5]) rdd2 = rdd.map(lambda x: x * 2) # 第一次Action，触发计算 print(rdd2.count()) # 输出5 # 第二次Action，又从头算一遍 print(rdd2.sum()) # 输出30

如果rdd2的计算代价很高，这种重复计算就很浪费。解决办法是用cache()或persist()把rdd2缓存起来，第一次Action后数据会留在内存里，后续Action直接读缓存。

3.3 cache与persist的选择策略

cache()和persist()都能缓存RDD，区别在于cache()默认用MEMORY_ONLY级别，persist()可以指定其他存储级别。

from pyspark import StorageLevel rdd = sc.parallelize(range(1000000)) rdd2 = rdd.map(lambda x: x * 2) # 方式一：cache，等价于persist(StorageLevel.MEMORY_ONLY) rdd2.cache() # 方式二：persist指定级别 rdd2.persist(StorageLevel.MEMORY_AND_DISK) # 触发计算并缓存 rdd2.count() # 后续操作直接读缓存 rdd2.sum()

存储级别的选择要看数据量和内存情况：

我个人的习惯是：数据量能放进内存就用MEMORY_ONLY，放不下就用MEMORY_AND_DISK。序列化级别虽然省内存，但每次读取都要反序列化，CPU开销不小，除非内存实在紧张，否则不太推荐。

注意：cache和persist是惰性的，调用后不会立即缓存，必须遇到第一个Action才会真正把数据存下来。另外记得在不需要缓存时调用unpersist()释放资源。

4. 常用Transformation算子逐个拆解

4.1 map、flatMap与filter：最基础的三板斧

map是最常用的转换算子，对RDD中每个元素应用一个函数，返回等长的新RDD。

rdd = sc.parallelize([1, 2, 3, 4, 5]) rdd2 = rdd.map(lambda x: x ** 2) print(rdd2.collect()) # [1, 4, 9, 16, 25]

flatMap跟map类似，但它的函数返回值是一个可迭代对象，flatMap会把这个可迭代对象“拍平”展开。

rdd = sc.parallelize(["hello world", "hello spark"]) rdd2 = rdd.flatMap(lambda line: line.split(" ")) print(rdd2.collect()) # ['hello', 'world', 'hello', 'spark']

map和flatMap的区别用一句话概括：map是“一对一”，flatMap是“一对多”。如果你处理的是嵌套结构，比如每行文本要拆成多个词，就用flatMap；如果是一对一转换，比如每个数字求平方，就用map。

filter按条件过滤元素，返回条件为True的元素组成的新RDD。

rdd = sc.parallelize([1, 2, 3, 4, 5, 6]) rdd2 = rdd.filter(lambda x: x % 2 == 0) print(rdd2.collect()) # [2, 4, 6]

这三个算子经常链式组合使用。比如统计一篇文章中长度超过5的单词：

rdd = sc.textFile("article.txt") result = rdd.flatMap(lambda line: line.split(" ")) \ .filter(lambda word: len(word) > 5) \ .map(lambda word: (word, 1)) \ .reduceByKey(lambda a, b: a + b)

4.2 reduceByKey与groupByKey：一个必须讲清楚的选择

这两个算子都涉及按key聚合，但性能差异巨大，是面试和实战中都绕不开的知识点。

reduceByKey会在每个分区内先做本地聚合（combine），再把结果shuffle到不同节点做全局聚合。

rdd = sc.parallelize([("a", 1), ("b", 2), ("a", 3), ("b", 4)]) rdd2 = rdd.reduceByKey(lambda x, y: x + y) print(rdd2.collect()) # [('a', 4), ('b', 6)]

groupByKey不做本地聚合，直接把所有相同key的value shuffle到同一个节点，然后聚成一个列表。

rdd = sc.parallelize([("a", 1), ("b", 2), ("a", 3), ("b", 4)]) rdd2 = rdd.groupByKey() print(rdd2.mapValues(list).collect()) # [('a', [1, 3]), ('b', [2, 4])]

两者的核心差异在于shuffle的数据量。假设有100万个("a", 1)的记录分布在10个分区，reduceByKey在每个分区先算出本地和，shuffle时每个分区只传一个("a", 本地和)，总共传10条记录。groupByKey则要把100万条记录全部shuffle，网络传输量差了十万倍。

所以原则很简单：能用reduceByKey就不用groupByKey。只有在确实需要拿到每个key对应的完整value列表时才用groupByKey，比如要计算每个key的中位数，那就没法用reduceByKey。

4.3 distinct、union与intersection：集合运算的代价

distinct去重，底层实现是先map成(key, None)，然后reduceByKey，最后取key。这个过程涉及shuffle，代价不低。

rdd = sc.parallelize([1, 2, 2, 3, 3, 3]) rdd2 = rdd.distinct() print(rdd2.collect()) # [1, 2, 3]

union合并两个RDD，不涉及shuffle，只是把两个RDD的分区简单拼接。

rdd1 = sc.parallelize([1, 2, 3]) rdd2 = sc.parallelize([4, 5, 6]) rdd3 = rdd1.union(rdd2) print(rdd3.collect()) # [1, 2, 3, 4, 5, 6]

intersection求交集，底层先做cogroup再过滤，涉及shuffle。

rdd1 = sc.parallelize([1, 2, 3, 4]) rdd2 = sc.parallelize([3, 4, 5, 6]) rdd3 = rdd1.intersection(rdd2) print(rdd3.collect()) # [3, 4]

这几个集合运算里，union最轻量，distinct和intersection都有shuffle开销。如果数据量很大，用之前先想想有没有替代方案。比如去重如果只是想去掉完全重复的行，可以在读取阶段用DataFrame的dropDuplicates，底层优化更好。

4.4 mapPartitions与mapPartitionsWithIndex：按分区批量处理

mapPartitions以分区为单位处理数据，函数接收一个迭代器，返回一个迭代器。相比map的逐元素处理，mapPartitions可以减少函数调用开销，适合需要初始化昂贵资源的场景。

def process_partition(iterator): # 假设这里需要建立数据库连接 # conn = create_connection() results = [] for item in iterator: results.append(item * 2) # conn.close() return iter(results) rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=2) rdd2 = rdd.mapPartitions(process_partition) print(rdd2.collect()) # [2, 4, 6, 8, 10]

mapPartitionsWithIndex在mapPartitions基础上多传一个分区索引参数，方便你知道当前处理的是第几个分区。

def process_with_index(index, iterator): for item in iterator: yield (index, item) rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=2) rdd2 = rdd.mapPartitionsWithIndex(process_with_index) print(rdd2.collect()) # [(0, 1), (0, 2), (1, 3), (1, 4), (1, 5)]

注意：mapPartitions的迭代器只能遍历一次。如果你在函数里遍历了两遍，第二遍会得到空结果。需要多次遍历时，先把迭代器转成列表。

5. Action算子与结果输出：什么时候真正开始算

5.1 collect、take与first：把数据拉回本地

collect把RDD所有元素拉回Driver端，返回一个Python列表。这是最常用的调试手段，但也是最危险的算子——如果数据量很大，Driver端内存直接爆掉。

rdd = sc.parallelize([1, 2, 3, 4, 5]) result = rdd.collect() print(result) # [1, 2, 3, 4, 5]

take(n)只取前n个元素，比collect安全得多。它不会扫描整个RDD，Spark只会计算包含前n个元素的分区，取够就停。

rdd = sc.parallelize(range(1000000)) result = rdd.take(5) print(result) # [0, 1, 2, 3, 4]

first等价于take(1)，返回第一个元素。

rdd = sc.parallelize([10, 20, 30]) print(rdd.first()) # 10

我个人的习惯是：调试阶段一律用take(10)或take(20)，绝不用collect。只有在确认结果集很小（比如聚合后的统计结果）时才用collect。

5.2 reduce与fold：聚合操作的细节差异

reduce对RDD中所有元素做聚合，函数必须满足交换律和结合律，因为Spark不保证元素的处理顺序。

rdd = sc.parallelize([1, 2, 3, 4, 5]) result = rdd.reduce(lambda x, y: x + y) print(result) # 15

fold跟reduce类似，但多一个初始值参数。注意fold的初始值会在每个分区都应用一次，所以如果初始值不是单位元，结果可能跟预期不同。

rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=2) # 每个分区都会用0做初始值，所以结果跟reduce一样 result = rdd.fold(0, lambda x, y: x + y) print(result) # 15 # 如果初始值不是0，结果会多出来 result = rdd.fold(10, lambda x, y: x + y) print(result) # 10 * 2 + 15 = 35

上面这个例子很能说明问题：fold(10)的结果是35而不是25，因为两个分区各加了一次10。这个坑我在实际项目中踩过，当时用fold做累加，初始值设了个非零数，结果怎么都对不上，排查了半天才发现是分区重复应用初始值的问题。

5.3 foreach与saveAsTextFile：把结果写出去

foreach对每个元素应用一个函数，通常用于写外部存储，比如写数据库。

def write_to_db(row): # 实际写数据库的逻辑 pass rdd = sc.parallelize([1, 2, 3]) rdd.foreach(write_to_db)

saveAsTextFile把RDD保存成文本文件，每个分区生成一个part文件。

rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=2) rdd.saveAsTextFile("output/rdd_result") # 输出目录下会有 part-00000 和 part-00001 两个文件

注意：saveAsTextFile的路径不能已存在，否则会报错。另外输出是目录不是单个文件，每个分区一个part文件。如果想让结果合并成一个文件，可以先repartition(1)再保存，但这样会损失并行度，数据量大时不建议。

6. 分区调优与数据倾斜：RDD性能的关键战场

6.1 如何判断分区数是否合理

分区数调优没有万能公式，但有几个判断依据。

第一，看任务执行时的并行度。在Spark UI的Stages页面，如果看到大量任务处于等待状态，说明分区数不够，集群资源没吃满。如果每个任务处理的数据量很小（几KB），任务调度开销占比过高，说明分区数太多。

第二，看每个分区的数据量。用rdd.glom().map(len).collect()可以查看每个分区的元素数量。理想情况下各分区数据量应该大致均衡，如果差异超过10倍，就有数据倾斜的风险。

第三，看shuffle后的分区数。reduceByKey、groupByKey等算子可以指定分区数：

rdd = sc.parallelize([("a", 1), ("b", 2), ("a", 3)]) # 指定shuffle后分区数为4 rdd2 = rdd.reduceByKey(lambda x, y: x + y, numPartitions=4) print(rdd2.getNumPartitions()) # 4

shuffle后的分区数默认等于父RDD的分区数，但通常需要调整。经验值是shuffle后每个分区处理128MB左右的数据。

6.2 repartition与coalesce：调整分区的两种方式

repartition可以增加或减少分区数，但一定会触发shuffle。

rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=2) rdd2 = rdd.repartition(4) print(rdd2.getNumPartitions()) # 4

coalesce只能减少分区数，默认不触发shuffle，只是把多个分区合并成一个。

rdd = sc.parallelize([1, 2, 3, 4, 5], numSlices=4) rdd2 = rdd.coalesce(2) print(rdd2.getNumPartitions()) # 2

选择原则：增加分区用repartition，减少分区优先用coalesce。如果减少分区后各分区数据量严重不均，可以给coalesce传shuffle=True，这样会触发shuffle来均衡数据。

6.3 数据倾斜的识别与缓解

数据倾斜是RDD处理中最头疼的问题之一。表现是大部分任务很快跑完，少数几个任务卡住不动。原因是某些key对应的数据量远大于其他key。

识别方法：在Spark UI里看每个任务的shuffle read size，如果最大值和平均值差距悬殊，基本就是倾斜了。

缓解手段有几种：

加盐打散：给倾斜的key加上随机前缀，先做一次聚合，再去掉前缀做第二次聚合。

import random rdd = sc.parallelize([("hot_key", 1)] * 1000 + [("cold_key", 1)] * 10) # 第一步：加随机前缀 def add_salt(item): key, value = item return (f"{key}_{random.randint(0, 9)}", value) salted = rdd.map(add_salt) # 第一步聚合 partial = salted.reduceByKey(lambda x, y: x + y) # 第二步：去掉前缀 def remove_salt(item): key, value = item return (key.split("_")[0], value) desalted = partial.map(remove_salt) # 第二步聚合 final = desalted.reduceByKey(lambda x, y: x + y) print(final.collect())

提高并行度：增加shuffle后的分区数，让倾斜的数据分散到更多任务里。这个方法治标不治本，但实现简单，很多时候够用。

过滤异常key：如果倾斜是由少数几个异常key造成的，而且这些key的业务价值不大，可以直接过滤掉。

提示：加盐打散会增加一次shuffle，代价不小。如果倾斜不严重，优先考虑提高并行度。只有倾斜非常严重且无法通过其他手段缓解时，才用加盐方案。

7. 几个我实际踩过的坑和对应的解法

7.1 闭包引用外部变量导致的序列化问题

在PySpark里，Transformation里的函数会被序列化后发送到各个Executor执行。如果函数引用了Driver端的对象，这个对象也必须能被序列化。

class MyProcessor: def __init__(self): self.factor = 2 def process(self, x): return x * self.factor processor = MyProcessor() rdd = sc.parallelize([1, 2, 3]) # 这样写会报序列化错误 # rdd2 = rdd.map(processor.process)

解决办法是把需要的属性提取成局部变量，或者用广播变量。

factor = 2 rdd2 = rdd.map(lambda x: x * factor)

如果确实需要传递大对象，用广播变量：

broadcast_factor = sc.broadcast({"factor": 2}) rdd2 = rdd.map(lambda x: x * broadcast_factor.value["factor"])

广播变量的好处是每个Executor只存一份，而不是每个任务一份，能省不少内存。

7.2 迭代器只能遍历一次

前面提过mapPartitions的迭代器只能遍历一次，这里展开说一下。很多人习惯先把迭代器转成列表再处理，但如果数据量大，列表会占大量内存，失去mapPartitions的意义。

# 不推荐：把整个分区数据加载到内存 def bad_process(iterator): data = list(iterator) total = sum(data) return [x / total for x in data] # 推荐：用生成器，边读边处理 def good_process(iterator): data = list(iterator) # 这里还是需要list，因为要算总和 total = sum(data) for x in data: yield x / total

如果确实需要多次遍历且数据量大，考虑用itertools.tee，但它也会缓存数据，本质上还是占内存。最好的办法是重新设计算法，避免多次遍历。

7.3 collect大结果集导致Driver OOM

这个坑我踩过不止一次。调试时随手一个collect，数据量小的时候没事，数据量一大Driver直接挂掉。

# 危险操作 # result = rdd.collect() # 安全替代方案 result = rdd.take(100) # 只取前100条 # 或者先聚合再collect result = rdd.reduceByKey(lambda x, y: x + y).collect() # 聚合后结果集小很多

如果确实需要把全量结果写到本地，用saveAsTextFile写到分布式存储，而不是collect回Driver。

7.4 频繁创建RDD导致性能下降

有些人写代码习惯每一步都创建一个新RDD变量，导致血缘链特别长。虽然Spark会做优化，但过长的血缘链在容错重算时代价很高。

# 不推荐：血缘链过长 rdd1 = sc.parallelize(range(1000)) rdd2 = rdd1.map(lambda x: x + 1) rdd3 = rdd2.filter(lambda x: x % 2 == 0) rdd4 = rdd3.map(lambda x: x * 2) rdd5 = rdd4.filter(lambda x: x > 100) # ... 继续链式操作 # 推荐：在关键节点checkpoint rdd1 = sc.parallelize(range(1000)) rdd2 = rdd1.map(lambda x: x + 1).filter(lambda x: x % 2 == 0) sc.setCheckpointDir("checkpoint_dir") rdd2.checkpoint() # 截断血缘链 rdd3 = rdd2.map(lambda x: x * 2).filter(lambda x: x > 100)

checkpoint会把RDD物化到可靠存储（通常是HDFS），同时截断血缘关系。后续如果出错，从checkpoint点重新算就行，不用回溯整条链。代价是写磁盘的开销，所以只在血缘链很长或计算代价很高的场景用。

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

RDD灵活但写起来啰嗦，DataFrame声明式但受限于schema。实际项目中两者经常混用。

适合用RDD的场景：数据格式不规整，需要自定义解析逻辑；需要底层控制分区和shuffle行为；处理非结构化数据比如纯文本、二进制。

适合用DataFrame的场景：数据结构规整，有明确schema；需要做SQL风格的聚合和join；需要利用Catalyst优化器做自动优化。

两者可以互相转换：

from pyspark.sql import SparkSession spark = SparkSession.builder.appName("example").getOrCreate() # RDD转DataFrame rdd = sc.parallelize([(1, "a"), (2, "b")]) df = spark.createDataFrame(rdd, ["id", "name"]) df.show() # DataFrame转RDD rdd2 = df.rdd print(rdd2.collect())

我个人的经验是：数据清洗阶段用RDD处理不规整的原始数据，清洗完转成DataFrame做后续分析。这样既发挥了RDD的灵活性，又享受了DataFrame的优化能力。

最后分享一个实用技巧：如果你不确定该用RDD还是DataFrame，先试着用DataFrame写，如果发现某些转换用DataFrame表达起来特别别扭，再局部换成RDD。不要一上来就全用RDD，那样会失去很多自动优化的机会。