23 钨丝计划：Tungsten给开发者带来了哪些福报？
你好，我是吴磊。
通过前两讲的学习，我们知道在Spark SQL这颗智能大脑中，“左脑”Catalyst优化器负责把查询语句最终转换成可执行的Physical Plan。但是，把Physical Plan直接丢给Spark去执行并不是最优的选择，最优的选择是把它交给“右脑”Tungsten再做一轮优化。
Tungsten又叫钨丝计划，它主要围绕内核引擎做了两方面的改进：数据结构设计和全阶段代码生成（WSCG，Whole Stage Code Generation）。
今天这一讲，我们就来说说Tungsten的设计初衷是什么，它的两方面改进到底解决了哪些问题，以及它给开发者到底带来了哪些性能红利。
Tungsten在数据结构方面的设计
相比Spark Core，Tungsten在数据结构方面做了两个比较大的改进，一个是紧凑的二进制格式Unsafe Row，另一个是内存页管理。我们一个一个来说。
Unsafe Row：二进制数据结构
Unsafe Row是一种字节数组，它可以用来存储下图所示Schema为（userID，name，age，gender）的用户数据条目。总的来说，所有字段都会按照Schema中的顺序安放在数组中。其中，定长字段的值会直接安插到字节中，而变长字段会先在Schema的相应位置插入偏移地址，再把字段长度和字段值存储到靠后的元素中。更详细的例子我们在第9讲说过，你可以去看看。
那么，这种存储方式有什么优点呢？我们不妨用逆向思维来思考这个问题，如果采用JVM传统的对象方式来存储相同Schema的数据条目会发生什么。
JVM至少需要6个对象才能存储一条用户数据。其中，GenericMutableRow用于封装一条数据，Array用于存储实际的数据值。Array中的每个元素都是一个对象，如承载整型的BoxedInteger、承载字符串的String等等。这样的存储方式有两个明显的缺点。
首先，存储开销大。我们拿类型是String的name来举例，如果一个用户的名字叫做“Mike”，它本应该只占用4个字节，但在JVM的对象存储中，“Mike”会消耗总共48个字节，其中包括12个字节的对象头信息、8字节的哈希编码、8字节的字段值存储和另外20个字节的其他开销。从4个字节到48个字节，存储开销可见一斑。
其次，在JVM堆内内存中，对象数越多垃圾回收效率越低。因此，一条数据记录用一个对象来封装是最好的。但是，我们从下图中可以看到，JVM需要至少6个对象才能存储一条数据记录。如果你的样本数是1千亿的话，这意味着JVM需要管理6千亿的对象，GC的压力就会陡然上升。
我们反过来再看UnsafeRow，字节数组的存储方式在消除存储开销的同时，仅用一个数组对象就能轻松完成一条数据的封装，显著降低GC压力，可以说是一举两得。由此可见，Unsafe Row带来的潜在性能收益还是相当可观的。不过，Tungsten并未止步于此，为了统一堆外与堆内内存的管理，同时进一步提升数据存储效率与GC效率，Tungsten还推出了基于内存页的内存管理模式。
基于内存页的内存管理
为了统一管理Off Heap和On Heap内存空间，Tungsten定义了统一的128位内存地址，简称Tungsten地址。Tungsten地址分为两部分：前64位预留给Java Object，后64位是偏移地址Offset。但是，同样是128位的Tungsten地址，Off Heap和On Heap两块内存空间在寻址方式上截然不同。
对于On Heap空间的Tungsten地址来说，前64位存储的是JVM堆内对象的引用或者说指针，后64位Offset存储的是数据在该对象内的偏移地址。而Off Heap空间则完全不同，在堆外的空间中，由于Spark是通过Java Unsafe API直接管理操作系统内存，不存在内存对象的概念，因此前64位存储的是null值，后64位则用于在堆外空间中直接寻址操作系统的内存空间。
显然，在Tungsten模式下，管理On Heap会比Off Heap更加复杂。这是因为，在On Heap内存空间寻址堆内数据必须经过两步：第一步，通过前64位的Object引用来定位JVM对象；第二步，结合Offset提供的偏移地址在堆内内存空间中找到所需的数据。
JVM对象地址与偏移地址的关系，就好比是数组的起始地址与数组元素偏移地址之间的关系。给定起始地址和偏移地址之后，系统就可以迅速地寻址到数据元素。因此，在上面的两个步骤中，如何通过Object引用来定位JVM对象就是关键了。接下来，我们就重点解释这个环节。
如上图所示，Tungsten使用一种叫做页表（Page Table）的数据结构，来记录从Object引用到JVM对象地址的映射。页表中记录的是一个又一个内存页（Memory Page），内存页实际上就是一个JVM对象而已。只要给定64位的Object引用，Tungsten就能通过页表轻松拿到JVM对象地址，从而完成寻址。
那么，Tungsten使用这种方式来管理内存有什么收益呢？我们不妨以常用的HashMap数据结构为例，来对比Java标准库（java.util.HashMap）和Tungsten模式下的HashMap。
Java标准库采用数组加链表的方式来实现HashMap，如上图所示，数组元素存储Hash code和链表头。链表节点存储3个元素，分别是Key引用、Value引用和下一个元素的地址。一般来说，如果面试官要求你实现一个HashMap，我们往往也会采用这种实现方式。
但是，这种实现方式会带来两个弊端。
首先是存储开销和GC负担比较大。结合上面的示意图我们不难发现，存储数据的对象值只占整个HashMap一半的存储空间，另外一半的存储空间用来存储引用和指针，这50%的存储开销还是蛮大的。而且我们发现，图中每一个Key、Value和链表元素都是JVM对象。假设，我们用HashMap来存储一百万条数据条目，那么JVM对象的数量至少是三百万。由于JVM的GC效率与对象数量成反比，因此java.util.HashMap的实现方式对于GC并不友好。
其次，在数据访问的过程中，标准库实现的HashMap容易降低CPU缓存命中率，进而降低CPU利用率。链表这种数据结构的特点是，对写入友好，但访问低效。用链表存储数据的方式确实很灵活，这让JVM可以充分利用零散的内存区域，提升内存利用率。但是，在对链表进行全量扫描的时候，这种零散的存储方式会引入大量的随机内存访问（Random Memory Access）。相比顺序访问，随机内存访问会大幅降低CPU cache命中率。
那么，针对以上几个弊端，Tungsten又是怎么解决的呢？我们从存储开销、GC效率和CPU cache命中率分别来看。
首先，Tungsten放弃了链表的实现方式，使用数组加内存页的方式来实现HashMap。数组中存储的元素是Hash code和Tungsten内存地址，也就是Object引用外加Offset的128位地址。Tungsten HashMap使用128位地址来寻址数据元素，相比java.util.HashMap大量的链表指针，在存储开销上更低。
其次，Tungsten HashMap的存储单元是内存页，内存页本质上是Java Object，一个内存页可以存储多个数据条目。因此，相比标准库中的HashMap，使用内存页大幅缩减了存储所需的对象数量。比如说，我们需要存储一百万条数据记录，标准库的HashMap至少需要三百万的JVM对象才能存下，而Tungsten HashMap可能只需要几个或是十几个内存页就能存下。对比下来，它们所需的JVM对象数量可以说是天壤之别，显然，Tungsten的实现方式对于GC更加友好。
再者，内存页本质上是JVM对象，其内部使用连续空间来存储数据，内存页加偏移量可以精准地定位到每一个数据元素。因此，在需要扫描HashMap全量数据的时候，得益于内存页中连续存储的方式，内存的访问方式从原来的随机访问变成了顺序读取（Sequential Access）。顺序内存访问会大幅提升CPU cache利用率，减少CPU中断，显著提升CPU利用率。
如何理解WSCG？
接下来，我们再说说WSCG。首先，WSCG到底是什么？这就要提到内存计算的第二层含义了，它指的是在同一个Stage内部，把多个RDD的compute函数捏合成一个，然后把这一个函数一次性地作用在输入数据上。不过，这种捏合方式采用的是迭代器嵌套的方式。例如，土豆工坊中对于Stage0的处理，也就是下图中的fuse函数。它仅仅是clean、slice、bake三个函数的嵌套，并没有真正融合为一个函数。
WSCG指的是基于同一Stage内操作符之间的调用关系，生成一份“手写代码”，真正把所有计算融合为一个统一的函数。
什么是火山迭代模型？
那么，我们真的有必要把三个函数体融合成一个函数，甚至生成一份“手写代码”吗？迭代器嵌套的函数调用难道还不够吗？坦白说，迭代器嵌套还真不够。原因在于，迭代器嵌套的计算模式会涉及两种操作，一个是内存数据的随机存取，另一个是虚函数调用（next）。这两种操作都会降低CPU的缓存命中率，影响CPU的工作效率。这么说比较抽象，我们来举个小例子。
假设，现在有一张市民表，我们要从中统计在北京的人数。对应的语法树非常简单，从左到右，分别是数据扫描、过滤、投影和聚合。语法树先是经过“左脑”Catalyst优化器转换为Physical Plan，然后交付执行。Tungsten出现以前，Spark在运行时采用火山迭代模型来执行计算。这里，咱们需要先简单地介绍一下火山迭代模型（Volcano Iteration Model，以下简称VI模型）。
VI模型这种计算模式依托AST语法树，对所有操作符（如过滤、投影）的计算进行了统一封装，所有操作符都要实现VI模型的迭代器抽象。简单来说就是，所有操作符都需要实现hasNext和next方法。因此，VI模型非常灵活、扩展能力很强，任何一个算子只要实现了迭代器抽象，都可以加入到语法树当中参与计算。另外，为了方便操作符之间的数据交换，VI模型对所有操作符的输出也做了统一的封装。
那么，如果上面的查询使用VI模型去执行计算的话，都需要经过哪些步骤呢？对于数据源中的每条数据条目，语法树当中的每个操作符都需要完成如下步骤：
- 从内存中读取父操作符的输出结果作为输入数据
- 调用hasNext、next方法，以操作符逻辑处理数据，如过滤、投影、聚合等等
- 将处理后的结果以统一的标准形式输出到内存，供下游算子消费
因此，任意两个操作符之间的交互都会涉及我们最开始说的两个步骤，也就是内存数据的随机存取和虚函数调用，而它们正是CPU有效利用率低下的始作俑者。
WSCG的优势是什么？
Tungsten引入WSCG机制，正是为了消除VI模型引入的计算开销。这是怎么做到的呢？接下来，咱们还是以市民表的查询为例，先来直观地感受一下WSCG的优势。
对于刚刚的查询语句，WSCG会结合AST语法树中不同算子的调用关系，生成如上图所示的“手写代码”。在这份手写代码中，我们把数据端到端的计算逻辑（过滤、投影、聚合）一次性地进行了实现。
这样一来，我们利用手写代码的实现方式不仅消除了操作符，也消除了操作符的虚函数调用，更没有不同算子之间的数据交换，计算逻辑完全是一次性地应用到数据上。而且，代码中的每一条指令都是明确的，可以顺序加载到CPU寄存器，源数据也可以顺序地加载到CPU的各级缓存中，从而大幅提升了CPU的工作效率。
当然，WSCG在运行时生成的代码和我们这里举例的手写代码在形式上还有差别。不过，这也并不影响我们对于WSCG特性和优势的理解。看到这里，你可能会问：“WSCG不就是运行时的代码重构吗？”没错，本质上，WSCG机制的工作过程就是基于一份“性能较差的代码”，在运行时动态地（On The Fly）重构出一份“性能更好的代码”。
WSCG是如何在运行时动态生成代码的？
问题来了，WSCG是怎么在运行时动态生成代码的呢？
我们还是以刚刚市民表的查询为例，语法树从左到右有Scan、Filter、Project和Aggregate4个节点。不过，因为Aggregate会引入Shuffle、切割Stage，所以这4个节点会产生两个Stage。又因为WSCG是在一个Stage内部生成手写代码，所以，我们把目光集中到前三个操作符Scan、Filter和Project构成的Stage。
上一讲中我们说了，Spark Plan在转换成Physical Plan之前，会应用一系列的Preparation Rules。这其中很重要的一环就是CollapseCodegenStages规则，它的作用正是尝试为每一个Stages生成“手写代码”。
总的来说，手写代码的生成过程分为两个步骤：
- 从父节点到子节点，递归调用doProduce，生成代码框架
- 从子节点到父节点，递归调用doConsume，向框架填充每一个操作符的运算逻辑
这么说比较抽象，咱们以上面的第一个Stage为例，来直观地看看这个代码生成的过程。
首先，在Stage顶端节点也就是Project之上，添加WholeStageCodeGen节点。WholeStageCodeGen节点通过调用doExecute来触发整个代码生成过程的计算。doExecute会递归调用子节点的doProduce函数，直到遇到Shuffle Boundary为止。这里，Shuffle Boundary指的是Shuffle边界，要么是数据源，要么是上一个Stage的输出。在叶子节点（也就是Scan）调用的doProduce函数会先把手写代码的框架生成出来，如图中右侧蓝色部分的代码。
然后，Scan中的doProduce会反向递归调用每个父节点的doConsume函数。不同操作符在执行doConsume函数的过程中，会把关系表达式转化成Java代码，然后把这份代码像做“完形填空”一样，嵌入到刚刚的代码框架里。比如图中橘黄色的doConsume生成的if语句，其中包含了判断地区是否为北京的条件，以及紫色的doConsume生成了获取必需字段userId的Java代码。
就这样，Tungsten利用CollapseCodegenStages规则，经过两层递归调用把Catalyst输出的Spark Plan加工成了一份“手写代码”，并把这份手写代码会交付给DAGScheduler。拿到代码之后，DAGScheduler再去协调自己的两个小弟TaskScheduler和SchedulerBackend，完成分布式任务调度。
小结
Tungsten是Spark SQL的“右脑”，掌握它的特性和优势对SparkSQL的性能调优来说至关重要。具体来说，我们可以从它对内核引擎的两方面改进入手：数据结构设计和WSCG。
在数据结构方面，我们要掌握Tungsten的两项改进。
首先，Tungsten设计了UnsafeRow二进制字节序列来取代JVM对象的存储方式。这不仅可以提升CPU的存储效率，还能减少存储数据记录所需的对象个数，从而改善GC效率。
其次，为了统一管理堆内与堆外内存，Tungsten设计了128位的内存地址，其中前64位存储Object引用，后64位为偏移地址。
在堆内内存的管理上，基于Tungsten内存地址和内存页的设计机制，相比标准库，Tungsten实现的数据结构（如HashMap）使用连续空间来存储数据条目，连续内存访问有利于提升CPU缓存命中率，从而提升CPU工作效率。由于内存页本质上是Java Object，内存页管理机制往往能够大幅削减存储数据所需的对象数量，因此对GC非常友好的。
对于Tungsten的WSCG，我们要掌握它的概念和优势。
首先，WSCG指的是基于同一Stage内操作符之间的调用关系，生成一份“手写代码”，来把所有计算融合为一个统一的函数。本质上，WSCG机制的工作过程，就是基于一份“性能较差的代码”，在运行时动态地重构出一份“性能更好的代码”。
更重要的是，“手写代码”解决了VI计算模型的两个核心痛点：操作符之间频繁的虚函数调用，以及操作符之间数据交换引入的内存随机访问。手写代码中的每一条指令都是明确的，可以顺序加载到CPU寄存器，源数据也可以顺序地加载到CPU的各级缓存中，因此，CPU的缓存命中率和工作效率都会得到大幅提升。
每日一练
- 针对排序操作，你认为Tungsten在数据结构方面有哪些改进呢？
- 你认为表达式代码生成（Expression Codegen）和全阶段代码生成（Whole Stage Codegen）有什么区别和联系呢？
期待在留言区看到你的思考和答案，我们下一讲见！
- Fendora范东_ 👍（18） 💬（4）
有个地方没理解 1.onheap内存寻址所说的内存页表，保存的是对象引用到jvm对象地址的映射，那它应该也是一个map结构，它是不是用下面所说的Tungsten.hashmap实现的？ 2.我理解不是，下面说的Tungsten.hashmap实例，我理解它是一个完整查询结构:要查某条数据，先计算key的hashcode，然后拿到128位内存地址，然后高64位(堆内对象引用)用于在「内存页表」中查询jvm对象地址，低64位用于在查到的jvm对象中进行偏移计算拿到具体某行数据。 3.如果2我没分析错，那「内存页表」是怎么实现的呢？
2021-05-05 - 斯盖丸 👍（13） 💬（1）
老师，tungsten内存页这块还是看得我很迷糊。我试着用我自己的语言复述下你看下对不对。 假设我有一张一百万行的表，用Tungsten内存，那就是会把这一百万分散到几十个内存页里去是吗，也就是一个内存页存了几万行？ 其次，这几十个内存页里前64位都存着key value的键值对(共一百万个)，后64位存偏移指针。其中key就是每一行的hash，row就是表的每一行对吗？ 最后，如果我想找id=3的那行数据(假设id唯一)，那就再去内存页的后64位找偏移量为3，也就是指针挪动3个单位，来确定最终要找的那一行是吗？ 感觉自己都不能自圆其说了，要是面试这么回答估计得挂，求老师帮忙看看错在哪里…
2021-06-10 - kingcall 👍（6） 💬（2）
哈哈，昨天看了一遍没懂，然后去补了点知识 1 java unsafe 2 虚拟内存管理 3 spark 官网关于Tungsten的介绍，今天又来了！ 回答：关于sort 为了更好的利用CPU 的多级缓存，Tungsten 做了关于类似pointer-key 作为元素的数组，从而避免在主存里面随机读取数据进行排序，从而可以更好的利用缓存，其实这就是Tungsten 的第二点，然而这一点老师没有介绍，估计是在这等着的吧，哈哈！
2021-05-07 - Sean 👍（4） 💬（1）
老师提到,开启了堆外之后，Spark在运行时会优先使用堆外，堆外不够再回退到堆内。我理解为这个任务一共生成了1000个task,每个task100m,堆外内存是6G,堆内2G,在执行到第50个task时,发现堆内还剩下40m,则剩下的所有task,960个都会走堆内内存,即时堆外50个task占用的内存已经释放,依然不会被使用,不知道这样理解对不对
2021-09-04 - keeprun 👍（2） 💬（1）
老师好，最近在看groupBy的Aggregation策略的选择，包含Hash-based Aggregation（spark 2.2.0后增加了Object-Hash-based Aggregation）和Sort-based Aggregation。其中能否使用Hash-based Aggregation的判断条件主要是UnsafeRow.isMutable(field.dataType())，主要是定长的数据类型，看到注释中提到，数据能够就地更新（Field types that can be updated in place in UnsafeRows）。是否主要是效率考虑（之前第9节有说明UnsafeRow的存储方式，定长的数据按顺序存储在字节数组中，而变长的字段需要通过offset来记录。）还是其他有其他原因？麻烦解惑。
2021-11-27 - wow_xiaodi 👍（2） 💬（1）
老师，有个问题，java的hashmap对于哈希冲突的元素可以通过遍历链表来定位到目标对象，那么tungsten.hashmap的value存放的却是一个128位内存地址，那么此时遇到哈希冲突，他是怎么解决的呢？是先根据128位的内容去寻址内存页的开始位置，然后一直遍历下去吗？
2021-08-12 - Stony.修行僧 👍（2） 💬（2）
学到很多，也参照了 《learn spark》，性能优化提高了不少，从好几个个小时job 优化到5分钟
2021-05-07 - zxk 👍（2） 💬（1）
问题二：Spark SQL 解析为语法树后，在不使用 Expression Codegen 的情况下，表达式节点每次执行都需要进行 Spark 内部的一些相关操作（如做一些操作类型匹配），那么 Spark 自身机制的开销可能大于我们需要执行的计算的开销，因此需要 Expression Codegen 对表达式进行代码生成，此时侧重于对表达式自身的优化；而 WSCG 则侧重与多个函数之间的合并，两者侧重点并不相同。 这里有几个疑问想请教下老师： 1. Tungsten 在堆内采用了 8 字节表示 Java Object，这跟 64 位 JVM 可以对应上，但 64 位 JVM 是有指针压缩机制的，这个对于 Tungsten 是否生效 2. Tungsten 在堆外有 64 位空间浪费了，为何 Spark 社区不针对堆内堆外区分处理，而是采用统一管理的方式？
2021-05-05 - 斯盖丸 👍（2） 💬（1）
老师，请问On heap寻址里的Object引用和偏移地址分别对应的是什么？Object引用是一条Row，偏移地址里是Row的一个字段或者说是列吗？
2021-05-05 - Unknown element 👍（1） 💬（1）
老师问下投影是什么？我看执行计划里好像就是选出需要的字段？
2021-11-04 - sparkjoy 👍（1） 💬（1）
老师，tungsten的hashmap我看应该在shufflehashjoin的时候会用到，broadcasthashjoin也用这个结构了么？
2021-10-12 - sparkjoy 👍（1） 💬（1）
老师，在内存页表中，怎么知道一个unsafe row的结束位置呢？
2021-08-25 - 快跑 👍（1） 💬（2）
1、内存页也是Java Object，所以内存页也存储在堆内内存On Heap? 内存页是一个连续内存空间，通过内存页+偏移量来定位一个数据元素；相比之前，每个数据元素都要生成对象，并且每个对象位置在内存中分散，属于随机访问 2、虽然Tungsten有管理 Off Heap 和 On Heap 内存空间，但是如果要使用Off Heap的情况，也是需要开启堆外内存spark.memory.offHeap.enabled=true，这个前提是没有改变的吧 3、开启堆外内存spark.memory.offHeap.enabled=true，所有的数据都存到堆外了么，还是有区分。
2021-05-14 - 西南偏北 👍（1） 💬（1）
花了一个五一的时间，从第一篇追上了老师的更新进度哈哈！老师厉害，学到了很多，尤其Spark SQL这块，还是要对着老师讲的东西，再去认真的读下源码才能有更深的理解
2021-05-05 - 科学养牛 👍（0） 💬（1）
没选过计算机原理，感觉这讲完全听不懂了😭
2021-05-12