(1)部署
部署方式: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程序的的部署、调优
文章位置:
http://bzwzjz.com/article/pdjhod.html