总结一下，避免后面再重复踩坑。

Spark Streaming是近实时(near real time)的小批处理系统， 可以对接各类消息中间或者直接监控Hdfs目录， 可以做为实时大数据流式计算，也可以做一些按时间窗口的数据聚合分析，比如流量监控之类的， 主要的优势是和spark-sql, spark-mlib, spark-graphx无缝结合的生态系统。

官方地址： http://spark.apache.org/docs/2.2.0/streaming-programming-guide.html

Spark Streaming

上游数据可以是Kafka, Flume, Hdfs或者是TCP Sockets；处理后的下游数据可以是落到HDFS, 数据库， 或者重新写回消息中间件，随意处理。

maven环境

<dependency>
 <groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.2.0</version>`
</dependency>

spark-streaming2.20适配的消息中间件

官方给了一些例子：

nc -lk 9999 同一台机器上socket， 端口9999

$ ./bin/run-example streaming.NetworkWordCount localhost 9999

Spark Streaming的优势在于：

• 能运行在100+的结点上，并达到秒级延迟(最小设置batch-time为500ms，再小就很容易task大量堆积)。

• 使用基于内存的Spark作为执行引擎，具有高效和容错的特性。

• 能集成Spark的批处理和交互查询。

• 为实现复杂的算法提供和批处理类似的简单接口

spark Streaming封装了kafka的高级接口： Kafka Integration Guide.

Spark Streaming

DStream是spark-streaming提供的一个抽象数据类型， 就是按时间切分的一组有序RDD集合。

关于更多的概念和方法参考官网教程， 这里总结一下使用的一些坑和优化：

一， kerberos 认证问题：

问题： 我们的hadoop访问有kerberos的认证机制，默认是7天更换，刚开始没注意这个问题，spark-streaming的程序每隔一周崩一次

解决：

1. –deploy-mode 由 yarn-client模式改为yarn-cluster模式；

2. –keytab /home/xxx/xxx.keytab –principal xxx@cloudera.xxx.com (刚开始客户端是2.1.0没生效，升级为2.2.0)

二， 优雅结束：

问题：application被人为中断，当前batch的数据没处理完

解决：源代码在spark.stop() 之前加了一个钩子， 来达到优雅退出， 保存断点checkpoint
–conf spark.streaming.stopGracefullyOnShutdown=true；

也可以自己在JVM关闭之前添加钩子， 来附加做一些邮件报警之类的事情(发送kill命令关闭driver进程，不要使用（－９）强制关闭，不然钩子无法捕获)

Runtime.getRuntime().addShutdownHook(

         new Thread() { override def run() {`

            log("Gracefully stop Spark Streaming")            `

              streamingContext.stop(true, true) } }`

      )

三， 数据缓存和清除：

cache或者persist的数据一定要在foreachRDD中清除掉，不然内存爆炸

spark.streaming.unpersist=true 这个配置只是自动推测并清除缓存数据， 最好还是代码中处理

四，batch的最大处理量，

根据内存和batchDuration设定合理的值, 保证batchDuration时间内能处理完，不造成堆积， 也和流数据大小有关。

– conf spark.streaming.kafka.maxRatePerPartition=1000

五， 应用程序失败自动重启次数， 和重试间隔

  --conf spark.yarn.maxAppAttempts=4
  --conf [spark.yarn.am](http://spark.yarn.am).attemptFailuresValidityInterval=1h

六，使用YARN Capacity Scheduler调度， 且提交到单独的Yarn队列

     --queue realtime_queue

七，开启spark推测执行

# 推测执行开启

spark.speculation                     true

# 检测周期

spark.speculation.interval 100

# 完成task的百分比时启动推测

spark.speculation.quantile 0.75

# 比其他的慢多少倍时启动推测

spark.speculation.multiplier 1.5

八， 避免单个任务阻塞：

spark.streaming.concurrentJobs=4

九，合理的batchDuration:

不要小于500ms, 太小，会积压数据， 太大，实时性不好

十，合理GC: 开启并行Mark-Sweep垃圾回收机制， 其它的参照JVM的调优，减少full-GC

–conf “spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC”

十一，计算效率：

实时计算对效率要求很高(不然大量任务堆积)， 所以spark的性能优化的方法在这里通用， 比如：

1. 合理的并行度partition， 一般是core的2~5倍, spark。 spark.default.parallelism=200

2. spark.sql.shuffle.partitions 设置大一点， 个人比较喜欢spark-sql处理逻辑，这个是sql shuffle时的并行度

3. spark.rdd.compress=true 缓存时压缩， 默认是false, 节省内存， 但是增加性能损耗

4. 参照 http://spark.apache.org/docs/latest/tuning.html

十二， 代码优化：

根据实际情况优化，在线任务和离线任务还是区别很大的，更多关注效率。

1. 处理空Batch:
   空batch比较多， 不判断直接写的话会形成很多空文件
   if(rdd.count() != 0) 或者 if(!rdd.partitions.isEmpty)
   推荐第二种， 数据量比较大时 count很费时间的

2. 高性能算子(平时要加强总结)：

   groupByKey  →  reduceByKey/aggregateByKey
   map →  mapPartitions
   foreachPartitions  →  foreach

1. 序列化(广播变量， cache, 自定义对象)：

通常图省事， 直接继承 java的Serializable 接口。

Spark支持使用Kryo序列化机制， 大概效率是java序列化的10倍， 变少网络传输的数据，减少在集群中耗费的内存资源。

• 配置参数： http://spark.apache.org/docs/latest/configuration.html

spark.serializer=org.apache.spark.serializer.KryoSerializer

spark.kryo.registrationRequired=true // 应用的类没有注册会报错，默认false

• 使用：需要先注册算子里边用到的类，不然会存储每个对象的全类名(full class name)，这样的使用方式往往比默认的 Java serialization 还要浪费更多的空间。

  • 需要序列化的类继承 java.io.Serializable
  • 注册类继承KryoRegistrato并且注册那些需要序列化的类
  • 在sparkConf中设置spark.serializer和spark.kryo.registrator

十三，其它

checkpoint: http://bit1129.iteye.com/blog/2217505 没用到自带的checkpoint机制

Kyro序列化

• Kryo使用: https://blog.csdn.net/cjuexuan/article/details/51485427

import com.esotericsoftware.kryo.Kryo
import org.apache.spark.serializer.KryoRegistrator

case class UserInfo(name: String ,age: Int,gender: String, addr: String)

class MyRegisterKryo extends KryoRegistrator {
  override def registerClasses(kryo: Kryo): Unit = {
    kryo.register(classOf[UserInfo])
  }
}

import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer
import scala.util.Random


/**
  * 需要序列化的类继承java.io.Serializable
  * 注册类继承KryoRegistrator并且注册那些需要序列化的类
  * 在sparkConf中设置spark.serializer和spark.kryo.registrator
  */

object KyroExample {

  def kyroExample() {
    val conf = new SparkConf().setMaster("local[1]").setAppName("KyroTest")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.kryo.registrator", "tools.MyRegisterKryo")
    conf.registerKryoClasses(Array(classOf[UserInfo], classOf[scala.collection.mutable.WrappedArray.ofRef[_]]))
    val sc = new SparkContext(conf)

    val arr = new ArrayBuffer[UserInfo]()

    val nameArr = Array[String]("lsw","yyy","lss")
    val genderArr = Array[String]("male","female")
    val addressArr = Array[String]("beijing","shanghai","shengzhen","wenzhou","hangzhou")

    for(i <- 1 to 1000){
      val name = nameArr(Random.nextInt(3))
      val age = Random.nextInt(100)
      val gender = genderArr(Random.nextInt(2))
      val address = addressArr(Random.nextInt(5))
      arr.+=(UserInfo(name,age,gender,address))
    }
    val start = System.currentTimeMillis()

    val rdd = sc.parallelize(arr)

    //序列化的方式将rdd存到内存
    rdd.persist(StorageLevel.MEMORY_ONLY_SER)
    println(System.currentTimeMillis() - start)
    sc.stop()
  }
}

• 用Kyro来读写硬盘文件 https://www.iteblog.com/archives/1328.html

 conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //使用Kryo序列化库

                  val sc = new SparkContext(conf)

 def saveAsObjectFile[T: ClassTag](rdd: RDD[T], path: String) {

                        val kryoSerializer = new KryoSerializer(rdd.context.getConf)    //  KryoSerializer对象， rdd.context.getConf获取缓存大小

rdd.mapPartitions(iter => iter.grouped(10)
      .map(_.toArray))
      .map(splitArray => {
      //initializes kyro and calls your registrator class
      val kryo = kryoSerializer.newKryo()   //map种创建Kryo实例, 线程不安全，只能放在map或者mappartition中
 
      //convert data to bytes
      val bao = new ByteArrayOutputStream()    
      val output = kryoSerializer.newKryoOutput()  
      output.setOutputStream(bao)
      kryo.writeClassAndObject(output, splitArray)
      output.close()
 
      // We are ignoring key field of sequence file
      val byteWritable = new BytesWritable(bao.toByteArray)
      (NullWritable.get(), byteWritable)
    }).saveAsSequenceFile(path)

}

def objectFile[T](sc: SparkContext, path: String, minPartitions: Int = 1)

    (implicit ct: ClassTag[T]) = {

    val kryoSerializer = new KryoSerializer(sc.getConf)

    sc.sequenceFile(path, classOf[NullWritable], classOf[BytesWritable],

       minPartitions)

       .flatMap(x => {

       val kryo = kryoSerializer.newKryo()

       val input = new Input()

       input.setBuffer(x._2.getBytes)

       val data = kryo.readClassAndObject(input)

       val dataObject = data.asInstanceOf[Array[T]]

       dataObject

    })

  }

参考：

Kryo读写硬盘: https://www.iteblog.com/archives/1328.html

Kryo使用: https://blog.csdn.net/cjuexuan/article/details/51485427