spark 数据序列化和内存调优（翻译）

由于大多数Spark计算的内存性质，Spark程序可能会被集群中的任何资源瓶颈：CPU、网络带宽或内存。大多数情况下，如果数据能放在内存，瓶颈是网络带宽，但有时，您还需要进行一些调整，例如以序列化形式存储RDD，以减少内存使用。本指南将涵盖两个主要主题：数据序列化，这对良好的网络性能至关重要，也可以减少内存使用，以及内存调整。

Data Serialization

序列化在任何分布式应用程序的性能中都起着重要的作用。将对象序列化很慢或消耗大量字节的格式会大大减慢计算速度。通常，这将是优化Spark应用程序的第一件事。Spark旨在在便利性（允许您在操作中使用任何Java类型）和性能之间取得平衡。它提供了两个序列化库：

• Java serialization:灵活但通常非常慢，并且导致许多类的序列化格式很大。
• Kryo serialization: Kryo 比 Java 序列化速度更快、更紧凑（通常快 10 倍），但不支持所有可序列化类型，并且需要您提前注册将在程序中使用的类以获得最佳性能。

Memory Tuning

调整内存使用有三个注意事项：对象使用的内存量（您可能希望整个数据集能放在内存）、访问这些对象的成本以及垃圾回收机制的开销（如果您的对象周转率很高）

默认情况下，Java 对象的访问速度很快，但很容易消耗比其字段内的“原始”数据多 2-5 倍的空间。这是由于以下几个原因：

• 每个不同的Java对象都有一个“对象头”，大约16个字节，包含指向其类的指针等信息。对于数据很少的对象（比如一个Int字段），这可能比数据大。
• Java字符串在原始字符串数据上有大约40个字节的开销（因为它们将其存储在Chars数组中并保留额外的数据，例如长度），并且由于String内部使用UTF-16编码，每个字符都存储为两个字节。因此，一个10个字符的字符串很容易消耗60个字节。
• 常见的集合类，如HashMap和LinkedList，使用链接数据结构，其中每个条目都有一个“包装器”对象（例如Map. Entry）。这个对象不仅有一个标题，还有指向列表中下一个对象的指针（通常每个8个字节）。
• 原始类型的集合通常将它们存储为“装箱”对象，例如java. lang.Integer。

Memory Management Overview

Spark中的内存使用在很大程度上属于以下两类之一：执行（execution）和存储(storage)。执行内存是指用于shuffles, joins, sorts and aggregations中的计算，而存储内存是指用于cache和跨集群传播内部数据的内存。在Spark中，执行和存储共享一个统一的内存区域（M）。当不使用执行内存时，存储可以获取所有可用内存，反之亦然。如果需要，执行可能会驱逐存储，但只有在总存储内存使用低于某个阈值（R）之前。换句话说，R描述了M中的一个子区域，其中缓存块永远不会被驱逐。由于实现的复杂性，存储可能不会驱逐执行。

这种设计确保了几个理想的属性。首先，不使用缓存的应用程序可以使用整个空间来执行，避免不必要的磁盘溢出。其次，使用缓存的应用程序可以保留最小的存储空间（R），其中它们的数据块不会被驱逐。最后，这种方法为各种工作负载提供了合理的开箱即用性能，而不需要用户了解如何在内部划分内存。

尽管有两个相关配置，但典型用户不需要调整它们，因为默认值适用于大多数工作负载：

• spark.memory.fraction M除以JVM堆空间（默认0.6）。其余空间（40%）保留给用户数据结构、Spark中的内部元数据，并在记录稀疏和异常大的情况下防止OOM错误。
• spark.memory.storageFraction R/M的值（默认 0.5）。 R 是 M 中的存储空间，其中缓存的块不会被执行驱逐。

Determining Memory Consumption

确定dataset所需内存消耗量的最佳方法是创建 RDD，将其放入缓存中，然后查看 Web UI 中的“Storage”页面。该页面会告诉您 RDD 占用了多少内存。

要估计特定对象的内存消耗，请使用

SizeEstimator

’s的估计方法。这对于尝试不同的数据布局来削减内存使用以及确定广播变量将在每个executor堆上占用的空间量很有用。

Tuning Data Structures

减少内存消耗的第一个方法是避免增加开销的 Java 功能，例如基于指针的数据结构和包装对象。做这件事有很多种方法：

1. 将您的数据结构设计为更倾向于对象数组和基本类型，而不是标准的Java或Scala集合类（例如HashMap）。Fastutil库为与Java标准库兼容的基本类型提供了方便的集合类。
2. 尽可能避免包含大量小对象和指针的嵌套结构。
3. 考虑使用数字 ID 或枚举对象而不是字符串作为键。
4. 如果您的 RAM 小于 32 GB，请设置 JVM 标志 -XX:+UseCompressedOops 以使指针为 4 个字节而不是 8 个字节。您可以在spark-env.sh中添加这些选项。

Garbage Collection Tuning

当您的程序存储的RDD有大量“churn”时，JVM垃圾回收机制可能会出现问题。（在只读取一次RDD然后对其运行许多操作的程序中，这通常不是问题。）当Java需要驱逐旧对象以为新对象腾出空间时，它需要跟踪所有Java对象并找到未使用的对象。这里要记住的要点是，垃圾收集的成本与 Java 对象的数量成正比，因此使用对象较少的数据结构（例如 Int 数组而不是 LinkedList）可以大大降低此成本。一个更好的方法是以序列化的形式持久化对象，如上所述：现在每个RDD分区只有一个对象（一个字节数组）。在尝试其他技术之前，如果GC有问题，首先要尝试的是使用序列化缓存。

由于task的工作内存（运行任务所需的内存）和节点上缓存的 RDD 之间的干扰，GC 也可能成为一个问题。我们将讨论如何控制分配给 RDD 缓存的空间来缓解这种情况。

Advanced GC Tuning

在Spark中进行GC调优的目的是确保只有长寿命RDD存储在老年代中，而年轻代的大小足以存储短寿命对象。这将有助于避免使用完整的GC来收集在任务执行期间创建的临时对象。一些可能有用的步骤是：

• 通过收集 GC 统计信息来检查是否存在过多垃圾回收。如果在任务完成之前多次调用完整GC，则意味着没有足够的内存可用于执行任务。
• 如果有太多的minor GC，但major GC不多，为Eden分配更多的内存会有所帮助。您可以将Eden的大小设置为对每个任务需要多少内存的高估。如果Eden的大小确定为E，那么您可以使用选项-Xmn=4/3*E设置年轻一代的大小。（扩大4/3也是为了考虑幸存者区域使用的空间。）
• 在打印的GC统计信息中，如果老年代接近满，则通过降低spark.memory.fraction来减少用于缓存的内存量；缓存更少的对象比减慢任务执行速度要好。或者，考虑减小年轻代的大小。如果您已按上述方式设置，这意味着降低-Xmn。如果没有，请尝试更改JVM的NewRatio参数的值。许多JVM将其默认为2，这意味着旧一代占据了堆的2/3。它应该足够大，以便该值超过spark.memory.fraction。
• 尝试使用-XX:+UseG1GC的G1GC垃圾回收器。它可以在垃圾回收机制成为瓶颈的某些情况下提高性能。请注意，对于大型执行器堆大小，使用-XX:G1HeapRegionSize增加G1区域大小可能很重要。
• 例如，如果您的任务正在从HDFS读取数据，则可以使用从HDFS读取的数据块的大小来估计任务使用的内存量。请注意，解压缩块的大小通常是块大小的2或3倍。因此，如果我们希望有3或4个任务的工作空间，并且HDFS块大小为128 MiB，我们可以估计Eden的大小为43128MiB。

Other Considerations

Level of Parallelism

除非您将每个操作的并行大小设置得足够高，否则集群将不会得到充分利用。Spark会根据每个文件的大小自动设置要在其上运行的“map”任务的数量（尽管您可以通过SparkContext. textFile等的可选参数来控制它），并且对于分布式“reduce”操作，例如groupByKey和elereByKey，它使用最大的父RDD的分区数。您可以将parallelism作为第二个参数传递（请参阅park.PairRDDFunctions留档），或者设置config属性park.default.并行来更改默认值。一般情况下，我们建议您的集群中每个CPU内核执行2-3个任务。

Data Locality

数据局部性会对Spark作业的性能产生重大影响。如果数据和对其进行操作的代码在一起，那么计算往往会很快。但是如果代码和数据是分开的，一个必须移动到另一个。通常，从一个地方到另一个地方运送序列化代码比运送一块数据要快，因为代码大小比数据小得多。Spark围绕数据局部性的一般原则构建了调度。

数据局部性是数据与处理它的代码的接近程度。根据数据的当前位置，有几个级别的局部性。从最近到最远的顺序：

• PROCESS_LOCAL 数据与正在运行的代码位于同一 JVM 中。这是最好的位置。
• NODE_LOCAL 数据位于同一节点上。例如，可能位于同一节点上的 HDFS 中，或者位于同一节点上的另一个executor中。这比 PROCESS_LOCAL 慢一点，因为数据必须在进程之间传输。
• NO_PREF 从任何地方都可以同样快速地访问数据，并且没有位置偏好。
• RACK_LOCAL 数据位于同一机架的服务器上。数据位于同一机架上的不同服务器上，因此需要通过网络发送，通常通过单个交换机。
• ANY 数据位于网络上的其他位置，而不是位于同一机架中。

Spark倾向于将所有任务安排在最佳locality级别，但这并不总是可能的。在任何空闲执行器上没有未处理数据的情况下，Spark会切换到较低的本地级别。有两种选择： a）等到繁忙的CPU空闲后再启动同一台服务器上处理该数据的任务 b）立即在更远的地方启动新任务，需要将数据移动到那里。

Spark通常会等待一段时间，希望忙碌的CPU能释放出来。一旦超时到期，它就会开始将数据从很远的地方移动到空闲的CPU。每个级别之间回退的等待超时可以单独配置，也可以在一个参数中一起配置；有关详细信息，请参阅配置页面上的spark.locality参数。如果您的任务执行耗时很长并且局部性较差，则应该增加这些设置，但默认值通常效果很好

Summary

这是一个简短的指南，指出了在调整Spark应用程序时应该了解的主要问题——最重要的是数据序列化和内存调整。对于大多数程序，切换到Kryo序列化并以序列化形式持久化数据将解决大多数常见的性能问题