如何调优Spark Steraming-阿里云开发者社区

云计算和大数据密不可分，这里有必要详细讨论下我的老本行——大数据领域。未来几年，我们将很荣幸地见证大数据技术的容器化。首先我们用几篇文章深入地了解一下大数据领域的相关技术。

1. 背景和简介

Spark Streaming是Spark的一个组件，它把流处理当作离散微批处理，被称为离散流或DStream。Spark的核心是RDD，即弹性分布式数据集。RDD本质上是将数据分区（Partition）封装起来。而DStream是一个由时间驱动、逻辑封装的RDD。以下面这段代码为例：

def main(args: Array[String]): Unit = {
    val logger = LoggerFactory.getLogger(PreJoin.getClass)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    val conf = new SparkConf().setMaster("local[2]").setAppName("PreJoin")
    //    val conf = new SparkConf().setAppName("PreJoin")
    /** 限制每秒钟从topic的每个partition最多消费的消息条数 */
    conf.set("spark.streaming.kafka.maxRatePerPartition", "500")
    conf.set("spark.streaming.kafka.consumer.cache.enabled", "false")
    // 设置每五秒更新一下
    val ssc = new StreamingContext(conf, Seconds(5))
    val sparkSession = SparkSession.builder().config(conf).getOrCreate()
    import sparkSession.implicits._
    var zookeeperservers = ""
     //  hbase的相关配置
    val config: Configuration = HBaseConfiguration.create()
    val newAPIJobConfiguration = Job.getInstance(config)
   // 读取Kafka
    val topics = Array("result1")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    // 读取另外两个topic
    // 每30秒一个滑动窗口处理30秒内到达的数据，将Json数据解析出来
    val resultJson = stream.map(record => record.value).window(Seconds(30), Seconds(30)).transform(rdd => {
      ...
    })
    // resultJson2和resultJson3类似
    val resultJson2 =
    val resultJson3 = 
    // 合并流
    val resultJsonAll = ssc.union(Seq(resultJson2, resultJson, resultJson3))
    // 遍历每个RDD，存入HBase
    resultJsonAll.foreachRDD(rdd => {
      val hbasePuts3 = data3.rdd.map((row: Row) => {
        ...
      })
      hbasePuts3.saveAsNewAPIHadoopDataset(newAPIJobConfiguration3.getConfiguration)
    })
    ssc.start() // 开始计算
    ssc.awaitTermination()
  } // 等待停止计算

上面这段代码描述了一个典型的Spark Streaming 的处理流程。它的功能是从Kafka拉取数据，经过一系列的转换，将结果存入HBase。我们可以看到流处理应用程序和批处理应用程序的一些区别。批处理应用程序拥有清晰的生命周期，它们一旦处理了输入文件就完成了执行。而上面的流处理应用程序的执行没有开始和停止的标记。

几个决定Spark Streaming应用程序生命周期的方法：

方法	描述
start()	开始执行应用程序
awaitTermination()	等待应用程序终止
stop()	强制应用程序停止执行

一个Spark应用程序的执行过程如下图

Yarn-Cluster运行模式执行过程

spark 控制进程

守护进程（Daemon）	描述
Driver（驱动程序）	包含SparkContext实例的应用程序入口点
Master（主进程）	负责调度和资源编排
Worker（子进程）	负责节点状态和运行执行器
Executor（执行器）	根据作业分配，负责执行该作业派发的任务

为了减少网络流量，强烈建议在集群机器上运行驱动程序，例如在Master节点，特别是需要驱动程序从Worker中提取数据的情况。

Spark分层执行结构

实体	描述
Application（应用程序）	SparkContext的一个实例
Job（作业）	一个Action后执行的一组阶段
Stage（阶段）	在shuffle内的一组转换
Task set（任务组）	来自同一组阶段的任务组
Task（任务）	一个阶段里的执行单元

有了上面的背景，我们下面便从几个方面来讨论下Spark Streaming的优化。

2. 调优

2.1 并行化

2.1.1 执行器Executor

num-executors

执行器是一个在每个Worker上执行的JVM进程。那么如何选择执行器的数量呢？理论上来说，既然executor是JVM进程，应该多一点才好。但是我们在选择executor数量的时候，有几条经验可供参考：

为每个节点上的操作系统和其他服务留出一些资源
如果在YARN上运行，也占用应用程序Master

executor-memory

该参数用于设置每个Executor进程的内存，Executor内存的大小，很多时候直接决定了Spark作业的性能。根据团队的资源队列的最大内存限制是多少， num-executors乘以 executor-memory，不能超过队列的最大内存量。建议申请的内存量最好不要超过资源队列最大总内存的1/3~1/2。

executor-cores

该参数置每个Executor进程的CPU core数量。这个参数决定了每个Executor进程并行执行task线程的能力。根据自己的资源队列的最大CPU core限制是多少，再依据设置的Executor数量，来决定每个Executor进程可以分配到几个CPU core。建议 num-executors * executor-cores不要超过队列总CPU core的1/3~1/2左右。

2.1.2 任务（Task）

Spark中的task是执行的单元。任务以线程而不是执行器的进程执行。每个DStream由RDD组成，而RDD又由分区组成。每个分区是一块独立的数据，由一个任务操作。因为一个RDD中的分区数与任务数之间存在几乎一对一的映射。也就是说，DStream并行度是分区数的函数。该分区数取决于依赖关系类型：跨过DStream时如果落在窄依赖区，分区数保持不变，但经过shuffle区由于宽依赖的缘故，这个分区数会发生改变。

shuffle的分区数由 spark.default.parallelism决定，或者如果 spark.default.parallelism未设置，则由构成父DStream的RDD中的最大分区数决定。

实现完全优化的并行度的最佳方法，就是不断试错，和常规Spark应用的调优的方法一样，控制逐渐增加分区的个数，每次将分区数乘以1.5，直到性能停止改进位置。这可以通过Spark UI 进行校准。

综上从Executor和Task的角度，得到Spark Streaming 的一些优化方法，提交Spark作业的脚本大概为：

./spark-submit \
  --master yarn \
  --num-executors 30 \
  --executor-memory 6G \
  --executor-cores 2 \
  --driver-memory 1G \
  --conf spark.default.parallelism=120 \
  --conf spark.storage.memoryFraction=0.5 \
  --conf spark.shuffle.memoryFraction=0.3 \
  --class com.example.PreJoin Stream-1.0-SNAPSHOT-jar-with-dependencies.jar

2.1.3 创建更多的输入DStream和Receive

每个输入DStream都会在某个Worker的Executor上启动一个Receiver，该Receiver接收一个数据流。因此可以通过创建多个DStream达到接收多个数据流的效果。

比如，一个接收多个Kafka Topic的输入DStream，可以拆分成多个输入DStream，每个分别接收一个topic的数据。这样就会创建多个Receiver，从而并行地接收数据，提高吞吐量。然后多个DStream使用union算子进行合并，从而形成一个DStream。核心代码如下：

// 合并流
val resultJsonAll = ssc.union(Seq(resultJson2, resultJson, resultJson3))

2.2 批处理间隔

批处理间隔是另一个直接影响性能的重要因素。它确定了微批处理的周期，也就是规定了每个微批处理能够通过的数据量。批处理间隔设置得太高则每个批处理会有高延迟，设置得太低则导致资源利用不足。

理想的状况是能够以线路速率处理数据，例如数据源每300毫秒发送一次，那么我们也可以这样假设：处理管道数据的延迟时间也为300毫秒。

如何设置批处理间隔，最好采取的策略是每次试验都从高值开始，比如1.5倍。Spark日志可用于计算系统的稳定性，即批处理间隔能否跟上数据速率。在日志中查找 Totaldelay总延迟。如果此值保持接近批处理间隔，则系统是稳定的。否则尝试增加2.1所述的并行化来减少管道的延迟。

假如日志如下：

JobScheduler: Total delay: 2.236 s for time 1581859740000 ms (execution: 2.256 s)

这表示该批处理的总延迟为3秒，小于批处理间隔。假设在某些时候可能会出现数据峰值，那么5秒是个不错的值。此外还可以通过Spark UI了解每阶段的延迟细目。Spark UI我们会在另一篇文章详细介绍。

2.3 内存

RDD基于内存计算，在内存中缓存所有内容，会给堆和垃圾收集器增加很大压力。如果应用程序的堆空间不足，可以增加 spark.executor.memory。此外有一些情况，Spark还会使用堆外内存，例如Java NIO采用的字节缓冲区。在YARN上，这个额外的内存分配由 spark.yarn.executor.memoryOverhead处理，默认值为 max(executorMemory*0.10384)。如果应用程序使用大量的堆外内存，那么应该增加这个因子。

一般来说，增加堆大小或堆外内存属于最后才会考虑的操作。我们首要的目标是减少应用程序的内存占用。下面介绍实现这一目标的三种方法。

2.3.1 序列化

RDD以序列化形式保存在内存中，可以减少内存使用并改善垃圾收集。默认情况下Spark使用Java序列化，这并不是很高效。Spark支持Kryo，Kryo更有效且性能高，可以将 spark.serializer设置为 org.apache.spark.serializer.KryoSerializer来启用Kryo。

2.3.2 压缩

除了序列化RDD之外。还可以将 spark.rdd.compress设置为true来进行压缩。

2.3.3 垃圾收集

流处理应用程序大量的对象增加了JVM垃圾收集的压力，频繁的GC会增加程序的延迟。建议对驱动程序和执行器使用CMS垃圾收集器，与应用程序同时运行垃圾收集来缩短暂停时间。

通过传递 --driver-java-options-XX:+UseConcMarkSweepGC到 spark-submit，为驱动程序启动CMS。对于执行器，将参数 spark.executor.extraJavaOptions设置为 XX:+UseConcMarkSweepGC，来启用CMS垃圾收集。

2.4 Shuffle

每次触发shuffle都会在集群中来回复制数据，这将付出很高的磁盘和网络I/O开销。因此在设计流应用程序的时候应该遵循一些原则：

2.4.1 提前投影过滤

提前进行投影和过滤，可以减少下游算子处理的数据。

2.4.2 多使用Combiner

Combiner使用的是map端聚合，可以减少在shuffle过程中需要处理的数据量。如使用reduceByKey(+)可以在shuffle之前的分区级别启用本地聚合。

2.4.2 大量运用并行化

shuffle操作内部使用分组操作的Hash映射来对分区空间进行分隔，这可能会导致堆空间耗尽。通过增加*ByKey()任务的的并行度，减少其工作集来避免这种情况。

2.4.3 文件合并

在大量shuffle任务的情况下，合并中间文件以改善磁盘查找是很有用的。可以设置 spark.shuffle.consolidateFiles为 true，启用合并。

2.4.4 更多内存

RDD，shuffle和应用程序对象之间共用执行器Java堆。默认情况下，RDD使用内存的60%（ spark.storage.memoryFraction），shuffle使用20%（ spark.shuffle.memoryFraction）。过多地使用将使shuffle聚合阶段的数据溢出到磁盘。如果使用shuffle比较多，则可以适当增加shuffle内存的的占用比例，以减少对磁盘的溢出次数。

Shuffle所涉及的问题比较复杂，调优的点也很多，我们会在另一篇文章详细介绍。

如何调优Spark Steraming