sparkStreaming程序的的部署、调优

发布时间:2020-09-20 06:51:53 来源:51CTO 阅读:1697 作者:原生zzy 栏目:大数据

(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分钟的数据,之后在处理完成之后,就会将数据进行清除

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

免费拨打  400 100 2938 免费拨打 400 100 2938
24小时售后技术支持 24小时售后技术支持
返回顶部 返回顶部