Spark Streaming之内存调优

待我称王封你为后i 2023-03-02 04:59 11阅读 0赞

## 一.内存调优 ##

Spark Streaming应用程序所需的集群内存大小在很大程度上取决于所使用的转换类型。例如，如果要对最后10分钟的数据使用窗口操作，则集群应具有足够的内存以将10分钟的数据保存在内存中。或者，如果想使用updateStateByKey操作大量的kv类型数据，则所需的存储空间会很大。相反，如果想执行一个简单的map-filter-store操作，则所需的内存将很少。

通常，由于通过接收器接收的数据存储在StorageLevel.MEMORY\_AND\_DISK\_SER\_2中，因此无法容纳在内存中的数据将溢出到磁盘上。这可能会降低流应用程序的性能，因此建议提供流应用程序所需的足够内存。最好尝试以小规模数据查看内存使用情况并据此进行估计。

内存调整的另一个方面是垃圾回收。对于需要低延迟的流应用程序，不希望由于JVM垃圾收集而导致较大的停顿。

有一些参数可以帮助调整内存使用和GC开销：

*  DStream的持久性级别：默认情况下，输入数据和RDD被持久化为序列化字节。与反序列化的持久性相比，这减少了内存使用和GC开销。启用Kryo序列化可进一步减少序列化的大小和内存使用量。通过压缩可以进一步减少内存使用量，但会占用CPU时间。
 *  清除旧数据：默认情况下，将自动清除DStream转换生成的所有输入数据和持久性RDD。Spark Streaming根据使用的转换来决定何时清除数据。例如，如果使用10分钟的窗口操作，那么Spark Streaming将保留最后10分钟的数据，并主动丢弃较旧的数据。通过设置streamingContext.remember可以将数据保留更长的时间。
 *  CMS垃圾收集器：强烈建议使用并发标记清除GC，以使与GC相关的暂停时间始终保持较低。尽管已知并发GC会降低系统的整体处理吞吐量，但仍建议使用它来实现更一致的批处理时间。确保在驱动程序【在spark-submit中添加配置–driver-java-options】和执行程序【使用Spark配置 spark.executor.extraJavaOptions】上都设置了CMS GC 。

为了进一步减少GC开销，可以尝试以下配置。

*  使用OFF\_HEAP存储级别持久化RDD 。
 *  使用更多具有较小堆大小的执行程序。这将减少每个JVM堆中的GC压力。

## 二.要点 ##

*  DStream与单个接收器关联。为了获得读取并行性，需要创建多个接收器，即多个DStream。接收器在执行器中运行。它占据了一个核心。订阅接收器topic后，请确保有足够的内核用于处理，即spark.cores.max应考虑接收器插槽。接收者以循环方式分配给执行者。
 *  当从流源接收数据时，接收器会创建数据块。每blockInterval【ms】生成一个新的数据块。在batchInterval期间创建了N个数据块，其中N = batchInterval / blockInterval。这些块由当前执行器的BlockManager分发给其他执行器的块管理器。之后，驱动程序上运行的网络输入跟踪器将被告知有关块的位置，以进行进一步处理。
 *  在驱动程序上为在batchInterval期间创建的块创建了RDD。在batchInterval期间生成的块是RDD的分区。每个分区都是一个任务。blockInterval == batchinterval表示将创建一个分区，并且可能在本地对其进行处理。
 *  块上的映射任务在执行器中进行处理【一个执行器接收该块，另一个执行器复制该块】，该执行器具有与块间隔无关的块，除非执行非本地调度。除非有更大的块间隔，这意味着更大的块。较高的spark.locality.wait值会增加在本地节点上处理块的机会。需要在这两个参数之间找到平衡，以确保较大的块在本地处理。
 *  无需依赖batchInterval和blockInterval，可以通过调用定义分区数inputDstream.repartition(n)。这会随机重新随机排列RDD中的数据以创建n个分区。是的，其可以获得更大的并行度。虽然以shuffle为代价。RDD的处理由driver的Jobscheduler安排为作业。在给定的时间点，只有一项作业处于活动状态。因此，如果一个作业正在执行，则其他作业将排队。
 *  如果有两个dstream，则将形成两个RDD，并且将创建两个作业，这些作业将一个接一个地调度。为避免这种情况，可以合并两个dstream。这将确保为dstream的两个RDD形成单个unionRDD。然后将此unionRDD视为一项工作。但是，RDD的分区不受影响。
 *  如果批处理时间超过batchinterval，那么显然接收方的内存将开始填满，并最终引发异常【最有可能是BlockNotFoundException】。当前，无法暂停接收器。使用SparkConf配置spark.streaming.receiver.maxRate，可以限制接收器的速率。