sparkStreaming程序的的部署、调优

标签：并行化   transform   速度   receiver   压缩   stage   提交   create   mat   

部署方式：spark standalone集群、yarn集群、meoss集群...
driver的HA：如果希望driver程序在失败之后自动重启，那么需要在程序中使用getOrcreate方法重构streamingContext对象，以及在spark-submit中添加参数。
Checkpoint目录的设置：如果程序使用到checkpoint目录，就必须配置一个hdfs兼容的文件系统作为checkpoint目录，因为程序是分布式的，不能将checkpoint单独的设置在某一个节点
接收数据的方式： Receiver和direct
  Receiver方式：需要给executor分配足够的资源，因为receiver接受的数据时存储在executor的内存中，尤其是在做window操作时，必须保证有足够的内存存储相应时间内的所有数据。 spark.streaming.receiver.writeAheadLog.enable参数设置为true，用于开启receive方式的WAL，保证数据不丢失。
  direct方式：对于kafka direct方式，引入了backpressure机制，从而不需要设置spark.streaming.kafka.maxRatePerPartition，spark可以自动的估计receive最合理的接收速度，并根据情况动态调整，只需要设置：spark.streaming.backpressure.enabled设置为true即可

（2）调优

• 设置合理的CPU资源
  • 在streaming程序中，CPU资源的使用可以分成两大类用于接收数据和用于处理数据，我们需要设置足够的CPU资源，使得有足够的CPU资源用于接收数据和处理数据，这样才能及时高效地处理数据。
• 数据接收的性能优化
  • 通过网络接收数据时，会将数据反序列化，并存储在spark的内存中。
  • 数据的并行化接收，就是启动多个receiver，设置多个DStream输入源
  • 调节block interval参数，对于大多数的receiver来说，在将接收的数据保存的时候，都会做数据切分成一个一个的block，而block的个数决定了每一个batch的分区个数，而分区个数则决定了transformation启动的task的个数：batch interval / block interval（spark.streaming.blockInterval，默认为200ms，最小值为50ms）
• Spark数据处理并行度调优
  • 如果在计算的任何stage中使用的并行task的数量没有足够多，那么集群的资源不能得到很好的利用，可以使用spark.default.parallelism，调整默认的并行的task个数，也可以在调用有shuffle的算子的时候手动指定numPar个数，调整task的并行度。
• Spark的任务调优
  如果每秒钟启动的task过于多，那么发送这些task去worker节点上的executor的性能开销就会比较大，此时延迟就会变得高了。
  • Task的序列化：使用Kryo序列化机制来序列化task，减少task的大小，从而减少发送到executor的时间
  • 执行模式，使用spark的自带的standalone运行spark程序，可以达到更少的task启动时间
• 序列化的调优
  • 输入数据，在receiver接收的数据时存储在executor内存中的，需要保证数据的0丢失从而序列化
  • 流式计算操作生成的持久化的RDD：流式计算操作生成的持久化的RDD，以及需要被窗口操作的数据都需要持久化
• batch interval调优
  • 对于流式计算而言，为了使其可以稳定运行以及高效运行，最重要的就是对于batch生成之后，就尽可能快的处理掉。 在构建StreamingContext的时候，需要我们传进一个参数，用于设置Spark Streaming批处理的时间间隔。Spark会每隔batchDuration时间去提交一次Job，如果你的Job处理的时间超过了batchDuration的设置，那么会导致后面的作业无法按时提交，随着时间的推移，越来越多的作业被拖延，最后导致整个Streaming作业被阻塞，这就间接地导致无法实时处理数据，最终导致程序崩溃。所以针对自己的业务，设置批处理时间尤为重要。
• Spark内存调优
  • DStream的持久化，将大量的数据持久化为byte数据，减少了数据序列化话后的对象，降低了GC的频率。当然为了进一步的降低内存的使用率，可以使用压缩：spark.rdd.compress这是为true即可
  • 清理旧数据，将保存在内存中已经被使用过的数据进行删除，释放内存。以window操作为例，如果窗口的时间为10分钟，spark中会保持这10分钟的数据，之后在处理完成之后，就会将数据进行清除

sparkStreaming程序的的部署、调优

标签：并行化   transform   速度   receiver   压缩   stage   提交   create   mat   

原文地址：http://blog.51cto.com/14048416/2339937