大数据系列(四) Spark：比MapReduce快100倍，它做了什么？

Spark比 MapReduce 快 100 倍它做了什么大数据系列第 4 篇MapReduce 的继任者来了内存计算到底香在哪先讲个真事儿2014 年DatabricksSpark 背后的公司搞了个比赛用 Spark 和 MapReduce 分别跑一个机器学习算法逻辑回归迭代 10 轮看谁能更快完成。结果出来了MapReduce110 秒Spark6 秒18 倍的差距。而且数据量越大、迭代轮数越多差距越夸张。PageRank 跑 10 轮Spark 比 MapReduce 快 30 倍。当时整个大数据圈都炸了。大家突然意识到原来 MapReduce 中间结果落盘的设计是个巨大的性能陷阱。Spark 的创始人 Matei Zaharia 说了一句话“我们不是在优化 MapReduce我们是在重新思考分布式计算。”今天咱们就来看看Spark 到底做了什么能让性能提升这么多。Spark 的核心思想数据放在内存里MapReduce 最大的性能瓶颈是什么每轮计算都要读写磁盘。MapReduce 的迭代计算 第 1 轮: HDFS → Map → Shuffle → Reduce → 写 HDFS 第 2 轮: 读 HDFS → Map → Shuffle → Reduce → 写 HDFS 第 3 轮: 读 HDFS → Map → Shuffle → Reduce → 写 HDFS ... 每轮都要重新读盘、写盘磁盘 I/O 成为瓶颈Spark 的想法很简单第一轮从 HDFS 读数据之后把中间结果放在内存里下一轮直接从内存读。Spark 的迭代计算 第 1 轮: HDFS → 计算 → 内存缓存 第 2 轮: 内存 → 计算 → 内存缓存 第 3 轮: 内存 → 计算 → 内存缓存 ... 后续轮次直接从内存读避免了磁盘 I/O内存读写速度是磁盘的 100-1000 倍。这就是 Spark 性能飞跃的根本原因。但把数据放内存说起来简单做起来可不容易。分布式环境下机器随时可能挂内存里的数据丢了怎么办Spark 的解决方案是RDD。RDD弹性分布式数据集RDDResilient Distributed Dataset是 Spark 最核心的抽象。咱们拆开来看这三个词分布式DistributedRDD 的数据分散在集群的多台机器上每个机器存一部分叫一个Partition。┌─────────────────────────────────────────────────────────────────┐ │ RDD 分区示意 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ 你的 RDD逻辑上是一个完整数据集 │ │ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ Partition 0 │ Partition 1 │ Partition 2 │ Partition 3 │ │ │ │ 机器 1 │ 机器 2 │ 机器 3 │ 机器 4 │ │ │ │ │ │ │ │ │ │ │ 数据子集 1 │ 数据子集 2 │ 数据子集 3 │ 数据子集 4 │ │ │ └─────────────────────────────────────────────────────────┘ │ │ │ │ 对 RDD 的操作 对每个 Partition 并行执行相同操作 │ │ │ └─────────────────────────────────────────────────────────────────┘弹性Resilient这是 RDD 最巧妙的设计。分布式环境下机器挂了是常态内存里的数据丢了怎么办Spark 的做法是不存数据本身只存数据从哪来、怎么算出来的。┌─────────────────────────────────────────────────────────────────┐ │ RDD 血统Lineage机制 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ HDFS 文件 │ │ │ │ │ ▼ textFile(hdfs://data.txt) │ │ RDD A ──→ flatMap(_.split( )) ──→ RDD B │ │ 拆分单词 │ │ │ │ │ ▼ map(word (word, 1)) │ │ RDD C ──→ reduceByKey(_ _) ──→ RDD D │ │ 按单词聚合 │ │ │ │ │ ▼ collect() │ │ 结果输出 │ │ │ │ 血统记录 │ │ • RDD B 来自 RDD A 的 flatMap 操作 │ │ • RDD C 来自 RDD B 的 map 操作 │ │ • RDD D 来自 RDD C 的 reduceByKey 操作 │ │ │ │ 容错场景 │ │ • 机器 3 挂了Partition 2 的数据丢了 │ │ • Spark 根据血统重新从 HDFS 读取对应数据 │ │ • 重新执行 flatMap → map → reduceByKey │ │ • 只重算丢失的分区不用重跑整个作业 │ │ │ └─────────────────────────────────────────────────────────────────┘RDD 会记录自己的血统Lineage——也就是我是从哪个 RDD 通过什么操作算出来的。如果某个分区的数据丢了Spark 根据血统信息重新计算这个分区就行。不需要像传统分布式系统那样全量复制数据来做备份。当然如果血统链太长比如几百轮迭代重算代价也很大。这时候可以用checkpoint()把 RDD 存到 HDFS切断血统链。数据集DatasetRDD 本质上就是一个不可变的、可分区的数据集合。你对 RDD 的操作不会修改它本身而是生成一个新的 RDD。RDD 的两种操作Transformation 和 ActionSpark 的操作分为两类这个区分很重要Transformation转换操作惰性求值Transformation 不会立即执行只是记录操作序列返回一个新的 RDD。// 这些操作都不会立即执行valrdd1sc.textFile(hdfs://data.txt)// 读取 HDFS 文件valrdd2rdd1.flatMap(_.split( ))// 拆分单词valrdd3rdd2.map(word(word,1))// 转换为 (word, 1)valrdd4rdd3.reduceByKey(__)// 按单词聚合// 到目前为止什么都没发生只是记录了要做这些事常见的 Transformation操作作用示例map()对每个元素做转换rdd.map(x x * 2)filter()按条件筛选rdd.filter(x x 10)flatMap()扁平化映射rdd.flatMap(_.split( ))reduceByKey()按 Key 聚合rdd.reduceByKey(_ _)groupByKey()按 Key 分组rdd.groupByKey()join()两个 RDD 关联rdd1.join(rdd2)distinct()去重rdd.distinct()Action行动操作触发执行遇到 ActionSpark 才会真正开始计算从血统的根节点一路算到当前 RDD。// 遇到 Action开始执行valresultrdd4.collect()// 把结果收集到 Driverprintln(result)// 其他 Actionrdd4.count()// 统计元素个数rdd4.take(10)// 取前 10 个rdd4.saveAsTextFile(hdfs://output)// 保存到 HDFSrdd4.foreach(println)// 遍历输出惰性求值的好处优化执行计划Spark 可以看到完整的操作链进行优化比如把多个 map 合并成一个避免不必要的计算如果某个 RDD 最终没被用到就不会执行DAG 调度比 MapReduce 聪明在哪MapReduce 的执行模型是固定的Map → Shuffle → Reduce。每个作业都要走这个流程哪怕你的计算逻辑很简单。Spark 的做法是把你的所有操作构建成一个DAG有向无环图然后智能地划分执行阶段。┌─────────────────────────────────────────────────────────────────┐ │ Spark DAG 与 Stage 划分 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ 你的代码 │ │ rdd.map(...).filter(...).reduceByKey(...).map(...).collect() │ │ │ │ 构建的 DAG │ │ │ │ HDFS ──→ [map] ──→ [filter] ──→ [reduceByKey] ──→ [map] ──→ 结果│ │ 窄依赖 窄依赖 宽依赖 窄依赖 │ │ │ │ Stage 划分 │ │ │ │ ┌─────────────────────────┐ ┌─────────────────┐ │ │ │ Stage 0 │ │ Stage 1 │ │ │ │ HDFS → map → filter │ │ reduceByKey │ │ │ │ │ │ → map → 结果 │ │ │ │ 窄依赖流水线执行 │ │ 宽依赖需要 │ │ │ │ │ │ Shuffle │ │ │ └─────────────────────────┘ └─────────────────┘ │ │ │ │ │ │ └────────── Shuffle ────────────┘ │ │ │ │ 窄依赖Narrow Dependency父 RDD 的每个分区只被子 RDD │ │ 的一个分区使用如 map/filter可以流水线执行 │ │ │ │ 宽依赖Wide Dependency父 RDD 的每个分区被子 RDD 的多个 │ │ 分区使用如 reduceByKey/groupByKey需要 Shuffle │ │ │ └─────────────────────────────────────────────────────────────────┘Spark 的聪明之处在于Stage 内流水线执行窄依赖的操作可以在同一个 Stage 内流水线执行不需要等待所有父分区完成减少不必要的 ShuffleMapReduce 每轮都要 ShuffleSpark 只在宽依赖处 Shuffle内存缓存中间结果Stage 之间的中间结果可以缓存在内存供后续复用缓存让数据住在内存里Spark 提供了persist()和cache()方法让 RDD 驻留在内存中valrddsc.textFile(hdfs://data.txt).flatMap(_.split( )).map(word(word,1))// 缓存到内存rdd.cache()// 等价于 persist(MEMORY_ONLY)// 第一次 Action 触发计算结果缓存到内存rdd.reduceByKey(__).collect()// 第二次使用同一个 RDD直接从内存读取不用重新计算rdd.filter(_._1.startsWith(a)).count()Spark 支持多种持久化级别级别存哪是否序列化特点MEMORY_ONLY内存否默认速度快但耗内存MEMORY_ONLY_SER内存是省内存但 CPU 要解压MEMORY_AND_DISK内存磁盘否内存不够时溢写到磁盘DISK_ONLY磁盘否内存极度有限时用序列化Serialization就是把对象转成字节数组可以大幅减少内存占用但读取时需要反序列化消耗 CPU。Spark SQL让 SQL 党也能用 SparkSpark 最初只有 RDD API需要写 Scala/Java 代码。后来推出了 Spark SQL可以用 SQL 或者 DataFrame API 操作数据。// 用 SQL 写 Spark 作业valdfspark.read.parquet(hdfs://users.parquet)df.createOrReplaceTempView(users)valresultspark.sql( SELECT age, COUNT(*) as cnt FROM users WHERE gender M GROUP BY age ORDER BY cnt DESC )result.show()Spark SQL 背后有个Catalyst 优化器会自动优化你的查询谓词下推把过滤条件推到数据源层减少数据读取量列裁剪只读需要的列不读整行常量折叠编译期计算常量表达式Whole-Stage Code Generation把多个操作编译成一段 Java 代码消除虚函数调用这些优化对开发者完全透明你写 SQLSpark 帮你优化到底。Spark Streaming微批次流处理Spark 也支持流处理但采用的是**微批次Micro-batch**模型实时数据流 ──────┬──────┬──────┬──────┬──────┬──────┬──────► │Batch1│Batch2│Batch3│Batch4│Batch5│... │ 1s │ 1s │ 1s │ 1s │ 1s │ ▼ ▼ ▼ ▼ ▼ ┌─────────────────────────────────────────┐ │ 每个批次当作一个小 RDD用批处理引擎处理 │ │ map → filter → reduceByKey → output │ └─────────────────────────────────────────┘把实时数据流切成一小段一小段的比如每 1 秒一个批次每个批次当作一个 RDD 来处理。优点和批处理共用同一套引擎和 API开发简单。缺点延迟是秒级的做不到毫秒级实时。如果你需要毫秒级的实时处理得看下一篇的 Flink。Spark 生态一览Spark 不只是一个计算引擎它发展成了一个完整的生态组件作用对标产品Spark Core核心计算引擎RDD-Spark SQL结构化数据处理Hive、PrestoSpark Streaming流处理微批次Storm、FlinkMLlib机器学习库Scikit-learnGraphX图计算Neo4j、GraphLab一套技术栈批处理、SQL 查询、流处理、机器学习、图计算都能干。这也是 Spark 受欢迎的原因之一——学习成本低换场景不用换技术栈。Spark vs MapReduce差距到底在哪维度MapReduceSpark中间结果写磁盘内存缓存迭代计算每轮重新读 HDFS从内存直接读任务启动每轮作业重新启动 JVMExecutor 长期运行编程语言主要 JavaScala/Java/Python/R交互式查询不支持Spark Shell 支持延迟分钟/小时级秒级内存计算容错任务重算血统重算 CheckpointSpark 的核心优势就三点内存计算避免中间结果落盘迭代性能提升 10-100 倍DAG 调度智能划分执行阶段减少不必要的 Shuffle统一生态批处理、SQL、流处理、机器学习一套搞定小结今天咱们聊了 Spark核心思想内存计算避免中间结果落盘RDD弹性分布式数据集血统机制实现高效容错Transformation vs Action惰性求值优化执行计划DAG 调度窄依赖流水线执行宽依赖才 ShuffleSpark SQLCatalyst 优化器自动优化查询Spark Streaming微批次模型秒级延迟Spark 的成功告诉我们有时候性能瓶颈不在算法而在架构设计。MapReduce 的中间结果落盘设计在当时是合理的内存贵、机器容易挂但随着硬件发展内存计算成为了更优的选择。你用过 Spark 吗是写 RDD 还是 DataFrame/SQL有没有遇到过内存溢出的 OOM 问题欢迎聊聊