1.Spark原理详解
2.Spark内存管理详解（下）——内存管理
3.spark sql源码系列 | with as 语句真的内内存会把查询的数据存内存嘛？
4.源码解析Spark中的Parquet高性能向量化读
5.spark原理系列 broadcast广播原理优缺点示例源码权威讲解
6.大数据面试题-Spark的内存模型

Spark原理详解

       Spark原理详解：

       Spark是一个专为大规模数据处理设计的内存计算框架，其高效得益于其核心组件——弹性数据分布集RDD。存源RDD是管理Spark的数据结构，它将数据存储在分布式内存中，内内存通过逻辑上的存源集中管理和物理上的分布式存储，提供了高效并行计算的管理汉服荟源码能力。

       RDD的内内存五个关键特性如下：

       每个RDD由多个partition组成，用户可以指定分区数量，存源默认为CPU核心数。管理每个partition独立处理，内内存便于并行计算。存源

       Spark的管理计算基于partition，算子作用于partition上，内内存无需保存中间结果，存源提高效率。管理

       RDD之间有依赖性，数据丢失时仅重新计算丢失分区，避免全量重算。

       对于key-value格式的RDD，有Partitioner决定分片和数据分布，优化数据处理的本地化。

       Spark根据数据位置调度任务，实现“移动计算”而非数据。

       Spark区分窄依赖（一对一）和宽依赖（一对多），前者不涉及shuffle，后者则会根据key进行数据切分。

       Spark的执行流程包括用户提交任务、生成DAG、划分stage和task、在worker节点执行计算等步骤。创建RDD的方式多样，包括程序中的集合、本地文件、HDFS、数据库、NoSQL和数据流等。

       技术栈方面，Spark与HDFS、YARN、MR、Hive等紧密集成，提供SparkCore、樱花飘落代码源码SparkSQL、SparkStreaming等扩展功能。

       在编写Spark代码时，首先创建SparkConf和SparkContext，然后操作RDD进行转换和应用Action，最后关闭SparkContext。理解底层机制有助于优化资源使用，如HDFS文件的split与partition关系。

       搭建Spark集群涉及上传、配置worker和master信息，以及启动和访问。内存管理则需注意Executor的off-heap和heap，以及Spark内存的分配和使用。

Spark内存管理详解（下）——内存管理

        弹性分布式数据集（RDD）作为Spark最根本的数据抽象，是只读的分区记录（Partition）的集合，只能基于在稳定物理存储中的数据集上创建，或者在其他已有的RDD上执行转换（Transformation）操作产生一个新的RDD。转换后的RDD与原始的RDD之间产生的依赖关系，构成了血统（Lineage）。凭借血统，Spark保证了每一个RDD都可以被重新恢复。但RDD的所有转换都是惰性的，即只有当一个返回结果给Driver的行动（Action）发生时，Spark才会创建任务读取RDD，然后真正触发转换的执行。

        Task在启动之初读取一个分区时，会先判断这个分区是否已经被持久化，如果没有则需要检查Checkpoint或按照血统重新计算。所以如果一个RDD上要执行多次行动，可以在第一次行动中使用persist或cache方法，在内存或磁盘中持久化或缓存这个RDD，从而在后面的行动时提升计算速度。事实上，cache方法是使用默认的MEMORY_ONLY的存储级别将RDD持久化到内存，故缓存是一种特殊的持久化。堆内和堆外存储内存的设计，便可以对缓存RDD时使用的内存做统一的规划和管理（存储内存的其他应用场景，如缓存broadcast数据，暂时不在本文的讨论范围之内）。

        RDD的持久化由Spark的Storage模块 [1] 负责，实现了RDD与物理存储的解耦合。Storage模块负责管理Spark在计算过程中产生的数据，将那些在内存或磁盘、在本地或远程存取数据的功能封装了起来。在具体实现时Driver端和Executor端的Storage模块构成了主从式的架构，即Driver端的BlockManager为Master，Executor端的BlockManager为Slave。Storage模块在逻辑上以Block为基本存储单位，RDD的每个Partition经过处理后唯一对应一个Block（BlockId的格式为 rdd_RDD-ID_PARTITION-ID ）。Master负责整个Spark应用程序的Block的元数据信息的管理和维护，而Slave需要将Block的更新等状态上报到Master，同时接收Master的命令，例如新增或删除一个RDD。

        在对RDD持久化时，Spark规定了MEMORY_ONLY、MEMORY_AND_DISK等7种不同的 存储级别 ，而存储级别是以下5个变量的组合 [2] ：

        通过对数据结构的分析，可以看出存储级别从三个维度定义了RDD的Partition（同时也就是Block）的存储方式：

        RDD在缓存到存储内存之前，Partition中的数据一般以迭代器（ Iterator ）的数据结构来访问，这是Scala语言中一种遍历数据集合的方法。通过Iterator可以获取分区中每一条序列化或者非序列化的数据项(Record)，这些Record的对象实例在逻辑上占用了JVM堆内内存的other部分的空间，同一Partition的不同Record的空间并不连续。

        RDD在缓存到存储内存之后，Partition被转换成Block，Record在堆内或堆外存储内存中占用一块连续的空间。将Partition由不连续的存储空间转换为连续存储空间的过程，Spark称之为“展开”（Unroll）。Block有序列化和非序列化两种存储格式，具体以哪种方式取决于该RDD的存储级别。非序列化的Block以一种DeserializedMemoryEntry的数据结构定义，用一个数组存储所有的Java对象，序列化的Block则以SerializedMemoryEntry的数据结构定义，用字节缓冲区（ByteBuffer）来存储二进制数据。每个Executor的Storage模块用一个链式Map结构（LinkedHashMap）来管理堆内和堆外存储内存中所有的Block对象的实例 [6] ，对这个LinkedHashMap新增和删除间接记录了内存的申请和释放。

        因为不能保证存储空间可以一次容纳Iterator中的所有数据，当前的计算任务在Unroll时要向MemoryManager申请足够的Unroll空间来临时占位，空间不足则Unroll失败，空间足够时可以继续进行。对于序列化的Partition，其所需的Unroll空间可以直接累加计算，一次申请。而非序列化的Partition则要在遍历Record的过程中依次申请，即每读取一条Record，采样估算其所需的Unroll空间并进行申请，空间不足时可以中断，释放已占用的Unroll空间。如果最终Unroll成功，当前Partition所占用的Unroll空间被转换为正常的缓存RDD的存储空间，如下图2所示。

        在 《Spark内存管理详解（上）——内存分配》 的图3和图5中可以看到，在静态内存管理时，Spark在存储内存中专门划分了一块Unroll空间，其大小是固定的，统一内存管理时则没有对Unroll空间进行特别区分，当存储空间不足是会根据动态占用机制进行处理。

        由于同一个Executor的所有的计算任务共享有限的存储内存空间，当有新的Block需要缓存但是剩余空间不足且无法动态占用时，就要对LinkedHashMap中的旧Block进行淘汰（Eviction)，而被淘汰的Block如果其存储级别中同时包含存储到磁盘的要求，则要对其进行落盘（Drop），否则直接删除该Block。

        存储内存的淘汰规则为：

        落盘的流程则比较简单，如果其存储级别符合 _useDisk 为true的条件，再根据其 _deserialized 判断是否是非序列化的形式，若是则对其进行序列化，最后将数据存储到磁盘，在Storage模块中更新其信息。

        Executor内运行的任务同样共享执行内存，Spark用一个HashMap结构保存了任务到内存耗费的映射。每个任务可占用的执行内存大小的范围为 1/2N ~ 1/N ，其中N为当前Executor内正在运行的任务的个数。每个任务在启动之时，要向MemoryManager请求申请最少为1/2N的执行内存，如果不能被满足要求则该任务被阻塞，直到有其他任务释放了足够的执行内存，该任务才可以被唤醒。

        执行内存主要用来存储任务在执行Shuffle时占用的内存，Shuffle是按照一定规则对RDD数据重新分区的过程，我们来看Shuffle的Write和Read两阶段对执行内存的使用：

        在ExternalSorter和Aggregator中，Spark会使用一种叫AppendOnlyMap的哈希表在堆内执行内存中存储数据，但在Shuffle过程中所有数据并不能都保存到该哈希表中，当这个哈希表占用的内存会进行周期性地采样估算，当其大到一定程度，无法再从MemoryManager申请到新的执行内存时，Spark就会将其全部内容存储到磁盘文件中，这个过程被称为溢存(Spill)，溢存到磁盘的文件最后会被归并(Merge)。

        Shuffle Write阶段中用到的Tungsten是Databricks公司提出的对Spark优化内存和CPU使用的计划 [4] ，解决了一些JVM在性能上的限制和弊端。Spark会根据Shuffle的情况来自动选择是否采用Tungsten排序。Tungsten采用的页式内存管理机制建立在MemoryManager之上，即Tungsten对执行内存的使用进行了一步的抽象，这样在Shuffle过程中无需关心数据具体存储在堆内还是堆外。每个内存页用一个MemoryBlock来定义，并用 Object obj 和 long offset 这两个变量统一标识一个内存页在系统内存中的地址。堆内的MemoryBlock是以long型数组的形式分配的内存，其 obj 的值为是这个数组的对象引用， offset 是long型数组的在JVM中的初始偏移地址，两者配合使用可以定位这个数组在堆内的绝对地址；堆外的MemoryBlock是直接申请到的内存块，其 obj 为null， offset 是这个内存块在系统内存中的位绝对地址。Spark用MemoryBlock巧妙地将堆内和堆外内存页统一抽象封装，并用页表(pageTable)管理每个Task申请到的内存页。

        Tungsten页式管理下的所有内存用位的逻辑地址表示，由页号和页内偏移量组成：

        有了统一的寻址方式，Spark可以用位逻辑地址的指针定位到堆内或堆外的内存，整个Shuffle Write排序的过程只需要对指针进行排序，并且无需反序列化，整个过程非常高效，对于内存访问效率和CPU使用效率带来了明显的提升 [5] 。

        Spark的存储内存和执行内存有着截然不同的管理方式：对于存储内存来说，Spark用一个LinkedHashMap来集中管理所有的Block，Block由需要缓存的RDD的Partition转化而成；而对于执行内存，Spark用AppendOnlyMap来存储Shuffle过程中的数据，在Tungsten排序中甚至抽象成为页式内存管理，开辟了全新的JVM内存管理机制。

        Spark的内存管理是一套复杂的机制，且Spark的版本更新比较快，笔者水平有限，难免有叙述不清、错误的地方，若读者有好的建议和更深的理解，还望不吝赐教。

spark sql源码系列 | with as 语句真的会把查询的数据存内存嘛？

       在探讨 Spark SQL 中 with...as 语句是否真的会把查询的数据存入内存之前，我们需要理清几个关键点。首先，网上诸多博客常常提及 with...as 语句会将数据存放于内存中，来提升性能。那么，实际情况究竟如何呢？

       让我们以 hive-sql 的视角来解答这一问题。在 hive 中，有一个名为 `hive.optimize.cte.materialize.threshold` 的参数。默认情况下，其值为 -1，代表关闭。当值大于 0 时（如设置为 2），with...as 语句生成的表将在被引用次数达到设定值后物化，从而确保 with...as 语句仅执行一次，进而提高效率。

       接下来，我们通过具体测试来验证上述结论。在不调整该参数的情况下，执行计划显示 test 表被读取了两次。此时，我们将参数调整为 `set hive.optimize.cte.materialize.threshold=1`，执行计划显示了 test 表被物化的情况，表明查询结果已被缓存。

       转而观察 Spark SQL 端，我们并未发现相关优化参数。Spark 对 with...as 的操作相对较少，在源码层面，汾酒溯源码查询通过获取元数据时所做的参数判断（如阈值与 cte 引用次数），我们可以发现 Spark 在这个逻辑上并未提供明确的优化机制，来专门针对 with...as 语句进行高效管理。

       综上所述，通过与 hive-sql 的对比以及深入源码分析，我们得出了 with...as 语句在 Spark SQL 中是否把数据存入内存的结论，答案并不是绝对的。关键在于是否通过参数调整来物化结果，以及 Spark 在自身框架层面并未提供特定优化策略来针对 with...as 语句进行内存管理。因此，正确使用 with...as 语句并结合具体业务场景，灵活调整优化参数策略，是实现性能提升的关键。

源码解析Spark中的Parquet高性能向量化读

       在Spark中，Parquet的高性能向量化读取是自2.0版本开始引入的特性。它与传统的逐行读取和解码不同，采用列式批处理方式，显著提升了列解码的速度，据Databricks测试，速度比非向量化版本快了9倍。本文将深入解析Spark的源码，揭示其如何支持向量化Parquet文件读取。

       Spark的向量化读取主要依赖于ColumnBatch和ColumnVector数据结构。ColumnBatch是每次读取返回的批量数据容器，其中包含一个ColumnVectors数组，每个ColumnVector负责存储一批数据中某一列的所有值。这种设计使得数据可以按列进行高效访问，同时也提供按行的视图，通过InternalRow对象逐行处理。

       在读取过程中，Spark通过VectorizedParquetRecordReader、VectorizedColumnReader和VectorizedValuesReader三个组件协同工作。VectorizedParquetRecordReader负责启动批量读取，它根据指定的批次大小和内存模式创建实例。VectorizedColumnReader和VectorizedValuesReader则负责实际的列值读取，根据列的类型和编码进行相应的解码处理。

       值得注意的是，Spark在数据加载时会重复使用ColumnBatch和ColumnVector实例，以减少内存占用，优化计算效率。ColumnVector支持堆内存和堆外内存，pubg怎么提取源码以适应不同的存储需求。通过这些优化，向量化读取在处理大型数据集时表现出色，尤其是在性能上。

       然而，尽管Spark的向量化读取已经非常高效，Iceberg中的Parquet向量化读取可能更快，这可能涉及到Iceberg对Parquet文件的特定优化，或者其在数据处理流程中的其他改进，但具体原因需要进一步深入分析才能揭示。

spark原理系列 broadcast广播原理优缺点示例源码权威讲解

       Spark广播（broadcast）的原理是通过将一个只读变量从驱动程序发送到集群上的所有工作节点，以便在运行任务时能够高效地访问这个变量。广播变量只会被发送一次，并且在工作节点上缓存，以供后续任务重用。

       这种方式可以避免在任务执行期间多次传输相同的数据，从而提高性能和效率。

       在Spark中，广播变量的实现主要依赖于DriverEndpoint和ExecutorEndpoint之间的通信机制。

       具体来说，当驱动程序将广播变量发送给工作节点时，它会使用BlockManager将序列化的块存储在内存中，并将块的元数据注册到BlockManagerMaster。

       然后，当工作节点执行任务时，它会向BlockManagerMaster请求获取广播变量的块，并从本地BlockManager中获取这些块的数据。这样，每个工作节点都可以在本地快速访问广播变量的数据。

       总结起来，Spark广播的实现涉及驱动程序对广播变量进行序列化和发送，以及工作节点接收、反序列化和缓存广播变量的块。这种机制有效地将只读数据分发到集群上的所有工作节点，提高了任务执行的性能和效率。

       广播变量在以下场景中非常有用：

       总之，广播变量适用于需要在多个任务之间共享只读数据，并且能够提供更高效的数据访问和减少网络传输开销的情况。通过使用广播变量，可以提高Spark应用程序的性能和效率。

       虽然广播在分布式计算中有很多优点，爱心代码源码获取但它也存在一些缺点：

       因此，在使用广播变量时需要考虑其局限性和适用场景。如果数据集较大，实时性要求高，或者需要频繁修改数据，可能需要考虑其他替代方案来避免广播的缺点。

       示例源码broadcast方法

       功能：将只读变量广播到集群,返回一个Broadcast对象以在分布式函数中进行读取变量将仅发送一次到每个执行器，同时调用了内部的方法broadcastInternal

       基础类Broadcast抽象类

       Broadcast 是 Spark 中的一个广播变量类。广播变量允许程序员在每台机器上缓存一个只读的变量，而不是将它与任务一起传输。通过使用广播变量，可以以高效的方式为每个节点提供大型输入数据集的副本。

       Broadcast 类的构造函数接收一个唯一标识符 id，用于标识广播变量。

       Broadcast 类是一个抽象类，有以下几个主要方法：

       Broadcast 类还定义了一些受保护的方法，用于实际获取广播变量的值、取消持久化广播变量的值以及销毁广播变量的状态。

       Broadcast 类还具有 _isValid 和 _destroySite 两个私有变量，分别表示广播变量是否有效（即尚未销毁）以及销毁广播变量的位置信息。

       总体来说，Broadcast 类提供了管理广播变量的功能，并确保广播变量的正确使用和销毁。

       实现类TorrentBroadcast

       TorrentBroadcast 是使用类似 BitTorrent 协议实现的 Broadcast 的具体实现（目前spark中只有一种实现）。它继承自 Broadcast 类，并提供以下功能：

       TorrentBroadcast 包含以下主要成员变量和方法：

       TorrentBroadcast 通过将广播数据分成小块并使用类似 BitTorrent 的协议进行分布式传输，以提高广播性能和可靠性。它允许在集群中高效地广播大量数据，并减少了驱动程序的负载。

       内部版本广播方法broadcastInternal

       该方法是spark内部版本的广播 - 将只读变量广播到集群，变量将仅发送一次到每个执行器。该方法中使用了broadcastManager对象中的newBroadcast创建广播变量

       broadcastManager初始化和创建广播对象初始化

       BroadcastManager构造函数会调用自身的initialize方法，创建一个TorrentBroadcastFactory实例.对象在实例化时，会自动调用自身的writeBlocks，把数据写入blockManager：

       使用了实现了BroadcastFactory接口的TorrentBroadcastFactory工厂方法。TorrentBroadcastFactory 是一个使用类似 BitTorrent 的协议来进行广播数据分布式传输的广播工厂。

       创建广播变量

       TorrentBroadcastFactory实例通过调用newBroadcast() 方法创建新的 TorrentBroadcast对象即广播变量。 可以参考上文实现类

       源码拓展BroadcastManager对象

       BroadcastManager 是 Spark 中负责管理广播变量的类。它包含以下主要功能：

       此外，BroadcastManager 还包含了一些内部变量，如下：

       总而言之，BroadcastManager 提供了广播变量的管理和操作功能，确保广播变量能够在集群中高效地分发和访问。

       BroadcastFactory接口

       BroadcastFactory 是 Spark 中所有广播实现的接口，用于允许多个广播实现。它定义了以下方法：

       通过实现BroadcastFactory 接口，可以自定义广播实现，并在 SparkContext 中使用相应的广播工厂来实例化广播变量。

       TorrentBroadcastFactory

       TorrentBroadcastFactory 是一个使用类似 BitTorrent 的协议来进行广播数据分布式传输的广播工厂。它实现了 BroadcastFactory 接口，并提供以下功能：

       TorrentBroadcastFactory 主要用于支持使用 BitTorrent-like 协议进行分布式传输的广播操作，以提高广播数据在集群中的传输效率和可靠性。

       BitTorrent 协议

       BitTorrent 是一种流行的文件分享协议，它使用了一种名为 "块链" 的技术。块链技术通常用于比特币等加密货币，但在 BitTorrent 中，它用于分发大型文件。

       BitTorrent 的工作原理

       初始化: 当一个用户想要下载一个文件时，他首先创建一个 "种子" 文件，这个文件包含该文件的所有块的哈希列表。 查找: 下载者使用 BitTorrent 客户端软件查找其他下载者，并请求他们分享文件块。 交换: 下载者与其他下载者交换文件块。每个下载者不仅下载文件，还同时通过上传已下载的块来帮助其他下载者。 完整性: 每个块都有一个哈希值，用于验证块的完整性。如果某个块的哈希值不匹配，则该块被认为是无效的，需要重新下载。

       块链技术

       BitTorrent 使用块链来确保每个块的完整性。每个块都包含前一个块的哈希值，这使得整个文件的所有块形成了一个链。如果某个块被修改或损坏，它的哈希值将不再匹配，BitTorrent 客户端将自动从其他下载者那里请求一个新的块。

       安全性

       BitTorrent 协议不使用加密，这意味着在交换文件块时，你的数据可能被第三方监听。为了提高安全性，你可以使用一个加密的 BitTorrent 客户端，如 BitTorrent Secure。

       总结

       BitTorrent 协议是一种高效的文件分享协议，它使用块链技术来保证文件块的完整性和安全性。然而，由于其不加密的特点，它可能不适合传输敏感信息。

大数据面试题-Spark的内存模型

       面试题来源：可回答：1）Spark内存管理的结构；2）Spark的Executor内存分布（参考“内存空间分配”）

       1、堆内和堆外内存规划

       作为一个JVM 进程，Executor 的内存管理建立在JVM的内存管理之上，Spark对JVM的堆内（On-heap）空间进行了更为详细的分配，以充分利用内存。同时，Spark引入了堆外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，进一步优化了内存的使用。

       堆内内存受到JVM统一管理，堆外内存是直接向操作系统进行内存的申请和释放。

       默认情况下，Spark 仅仅使用了堆内内存。Executor 端的堆内内存区域大致可以分为以下四大块：堆内内存的大小，由Spark应用程序启动时的 –executor-memory 或 spark.executor.memory 参数配置。这些任务在缓存 RDD 数据和广播（Broadcast）数据时占用的内存被规划为存储（Storage）内存，而这些任务在执行 Shuffle 时占用的内存被规划为执行（Execution）内存，剩余的部分不做特殊规划。

       Spark对堆内内存的管理是一种逻辑上的”规划式”的管理。不同管理模式下，这三部分占用的空间大小各不相同。

       堆外内存意味着把内存对象分配在Java虚拟机的堆以外的内存，这些内存直接受操作系统管理。这样做的结果就是能保持一个较小的堆，以减少垃圾收集对应用的影响。

       利用JDK Unsafe API（从Spark 2.0开始），Spark 可以直接操作系统堆外内存，减少了不必要的内存开销，以及频繁的 GC 扫描和回收，提升了处理性能。堆外内存可以被精确地申请和释放，而且序列化的数据占用的空间可以被精确计算，所以相比堆内内存来说降低了管理的难度，也降低了误差。

       在默认情况下堆外内存并不启用，可通过配置 spark.memory.offHeap.enabled 参数启用，并由 spark.memory.offHeap.size 参数设定堆外空间的大小。除了没有 other 空间，堆外内存与堆内内存的划分方式相同，所有运行中的并发任务共享存储内存和执行内存。

       2、内存空间分配

       静态内存管理与统一内存管理的区别在于存储内存和执行内存共享同一块空间，可以动态占用对方的空闲区域。

       统一内存管理的堆内内存结构如图所示：其中最重要的优化在于动态占用机制。统一内存管理的堆外内存结构如下图所示。

       凭借统一内存管理机制，Spark 在一定程度上提高了堆内和堆外内存资源的利用率，降低了开发者维护 Spark 内存的难度，但并不意味着开发者可以高枕无忧。如果存储内存的空间太大或者说缓存的数据过多，反而会导致频繁的全量垃圾回收，降低任务执行时的性能。

       3、存储内存管理

       RDD的持久化机制

       弹性分布式数据集（RDD）作为 Spark 最根本的数据抽象，是只读的分区记录（Partition）的集合。RDD的持久化由 Spark的Storage模块负责，实现了RDD与物理存储的解耦合。Storage模块负责管理Spark在计算过程中产生的数据，将那些在内存或磁盘、在本地或远程存取数据的功能封装了起来。在具体实现时Driver端和 Executor 端的Storage模块构成了主从式的架构。

       在对RDD持久化时，Spark规定了MEMORY_ONLY、MEMORY_AND_DISK 等7种不同的存储级别，而存储级别是以下5个变量的组合。

       通过对数据结构的分析，可以看出存储级别从三个维度定义了RDD的 Partition（同时也就是Block）的存储方式。

       4、执行内存管理

       执行内存主要用来存储任务在执行Shuffle时占用的内存。

       若在map端选择普通的排序方式，会采用ExternalSorter进行外排，在内存中存储数据时主要占用堆内执行空间。

       若在map端选择 Tungsten 的排序方式，则采用ShuffleExternalSorter直接对以序列化形式存储的数据排序，在内存中存储数据时可以占用堆外或堆内执行空间，取决于用户是否开启了堆外内存以及堆外执行内存是否足够。

       在Shuffle Write 阶段中用到的Tungsten是Databricks公司提出的对Spark优化内存和CPU使用的计划。在Shuffle过程中，Spark会根据Shuffle的情况来自动选择是否采用Tungsten排序。

       Tungsten 采用的页式内存管理机制建立在MemoryManager之上，即 Tungsten 对执行内存的使用进行了一步的抽象，这样在 Shuffle 过程中无需关心数据具体存储在堆内还是堆外。每个内存页用一个MemoryBlock来定义，并用 Object obj 和 long offset 这两个变量统一标识一个内存页在系统内存中的地址。

Spark原理 | 内存管理

       Spark作为一个基于内存的分布式计算引擎，其内存管理模块在整个系统中扮演着非常重要的角色。

        在执行Spark的应用程序时，Spark集群会启动Driver和Executor两种JVM进程：

        Spark管理的内存主要划分为4个区域：

        Executor作为一个JVM进程，它的内存管理建立在JVM的内存管理之上，Spark对JVM的堆内（On-heap）空间进行了更为详细的分配，以充分利用内存。同时，Spark引入了堆外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，进一步优化了内存的使用。

        堆内内存的大小，由 Spark 应用程序启动时的 executor-memory 或 spark.executor.memory 参数配置。Executor 内运行的并发任务共享 JVM 堆内内存，这些任务在缓存 RDD 数据和广播（Broadcast）数据时占用的内存被规划为存储（Storage）内存，而这些任务在执行 Shuffle 时占用的内存被规划为执行（Execution）内存，剩余的部分不做特殊规划，那些 Spark 内部的对象实例，或者用户定义的 Spark 应用程序中的对象实例，均占用剩余的空间。不同的管理模式下，这三部分占用的空间大小各不相同。

        Spark 对堆内内存的管理是一种逻辑上的"规划式"的管理，因为对象实例占用内存的申请和释放都由 JVM 完成，Spark 只能在申请后和释放前记录这些内存，我们来看其具体流程：

        为了进一步优化内存的使用以及提高 Shuffle 时排序的效率，Spark 引入了堆外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，存储经过序列化的二进制数据。利用 JDK Unsafe API（从 Spark 2.0 开始），在管理堆外的存储内存时不再基于 Tachyon，而是与堆外的执行内存一样，基于 JDK Unsafe API 实现，Spark 可以直接操作系统堆外内存，减少了不必要的内存开销，以及频繁的 GC 扫描和回收，提升了处理性能。堆外内存可以被精确地申请和释放，而且序列化的数据占用的空间可以被精确计算，所以相比堆内内存来说降低了管理的难度，也降低了误差。

        在默认情况下堆外内存并不启用，可通过配置 spark.memory.offHeap.enabled 参数启用，并由 spark.memory.offHeap.size 参数设定堆外空间的大小。除了没有 other 空间，堆外内存与堆内内存的划分方式相同，所有运行中的并发任务共享存储内存和执行内存。

        Spark 1.6 之后默认为统一管理（UnifiedMemoryManager）方式，1.6 之前采用的静态管理（StaticMemoryManager）方式仍被保留，可通过配置 spark.memory.useLegacyMode=true 参数启用静态内存管理方式。下面我们介绍下两种内存管理模型的进化。

        在 Spark 最初采用的静态内存管理机制下，存储内存、执行内存和其他内存的大小在 Spark 应用程序运行期间均为固定的，但用户可以应用程序启动前进行配置，堆内内存的分配如下所示：

        Spark 1.6 之后引入的统一内存管理机制，与静态内存管理的区别在于存储内存和执行内存共享同一块空间，可以动态占用对方的空闲区域。如下图所示：

        其中最重要的优化在于动态占用机制，其规则如下：

        新的版本引入了新的配置项：

        凭借统一内存管理机制，Spark 在一定程度上提高了堆内和堆外内存资源的利用率，降低了开发者维护 Spark 内存的难度，但并不意味着开发者可以高枕无忧。譬如，所以如果存储内存的空间太大或者说缓存的数据过多，反而会导致频繁的全量垃圾回收，降低任务执行时的性能，因为缓存的 RDD 数据通常都是长期驻留内存的。所以要想充分发挥 Spark 的性能，需要开发者进一步了解存储内存和执行内存各自的管理方式和实现原理。

Spark中cache和persist的区别

       cache

       ã€€ã€€é»˜è®¤æ˜¯å°†æ•°æ®å­˜æ”¾åˆ°å†…存中，懒执行

       ã€€ã€€def cache(): this.type = persist()

       ã€€ã€€persist

       ã€€ã€€å¯ä»¥æŒ‡å®šæŒä¹…化的级别。

       ã€€ã€€æœ€å¸¸ç”¨çš„是MEMORY_ONLY和MEMORY_AND_DISK。

       ã€€ã€€â€_2”表示有副本数。尽量避免使用_2和DISK_ONLY级别

       ã€€ã€€cache和persist的注意点

       ã€€ã€€1.都是懒执行(有的叫延迟执行)，需要action触发执行，最小单位是partition

       ã€€ã€€2.对一个RDD进行cache或者persist之后，下次直接使用这个变量，就是使用持久化的数据

       ã€€ã€€3.如果使用第二种方式，不能紧跟action算子

[SPARK][SQL] Tungsten Codegen优势与表达式生成

       今天，我们将深入探讨Spark中的CodeGen，特别是在Tungsten框架下的优势与表达式生成。

       Tungsten，启动于年，旨在优化CPU和内存使用。在对数据处理瓶颈进行分析后，发现主要瓶颈并非来自I/O或网络问题，而是CPU和内存限制。Tungsten通过代码生成，利用现代编译器和CPU的性能，优化了数据处理效率。

       Tungsten的CodeGen分为两个主要部分：基本表达式代码生成与全阶段代码生成（WSCG）。基本表达式代码生成针对的是最简单的运算任务，而WSCG则进一步整合同一Stage内的操作符，生成统一的代码块，将所有计算融合为一个函数。

       在Spark中，优化主要分为静态优化和动态优化。静态优化包括基于规则和基于代价的优化，目标是在执行前确定最优执行计划。动态优化，则通过自适应执行优化（AQE）等技术，在执行过程中调整计划，消除数据倾斜、降低IO，提高资源利用率。

       Spark Codegen的优势主要体现在两个方面：基本表达式的代码生成与全阶段代码生成。基本代码生成将表达式转化为独立的处理模块；全阶段代码生成则将多个RDD的计算整合，生成一次性执行的代码，大幅提升计算效率。

       与传统SQL执行的火山模型相比，Codegen的优势主要体现在：无虚函数调用，内存中的中间数据与CPU寄存器操作，以及循环展开与SIMD优化。在火山模型中，每次操作符处理数据需要保存在内存中，而Codegen则将数据直接放入寄存器，减少内存访问延迟。同时，编译器和CPU能够高效处理简单循环，避免了复杂函数调用带来的开销。

       Codegen的实现依赖于Janino动态编译器，该工具可以快速编译Java代码。在Codegen中，表达式被转换为特定的Java代码，省去了对象创建与函数调用的开销，提高了计算效率。

       在使用Janino实现表达式生成时，Spark定义了CodegenContext来管理生成代码所需的信息，如变量、函数和对象定义。表达式代码生成的过程涉及到将表达式绑定到输入模式、规范表达式、并生成实际的Java代码。

       对于Project（投影）操作的代码生成，Spark在关闭全阶段代码生成时，会调用特定的子类来生成Java代码，实现高效的数据投影逻辑。Project操作涉及到将输出列与输入模式绑定，生成Java代码以实现数据的投影与输出。

       总结而言，Tungsten Codegen通过静态和动态优化策略，显著提升了Spark的计算效率。它通过基本表达式代码生成与全阶段代码生成，有效减少了虚函数调用，优化了内存访问，以及利用SIMD特性，实现了高性能的数据处理。