第17课:Spark Streaming资源动态申请和动态控制消费速率原理剖析

本期内容:

  • Spark Streaming资源动态分配
  • Spark Streaming动态控制消费速率

为什么需要动态?

  • Spark默认情况下粗粒度的,先分配好资源再计算。而Spark Streaming有高峰值和低峰值,但是他们需要的资源是不一样的,如果按照高峰值的角度的话,就会有大量的资源浪费。
  • Spark Streaming不断的运行,对资源消耗和管理也是我们要考虑的因素。
  • Spark Streaming资源动态调整的时候会面临挑战:
  • Spark Streaming是按照Batch Duration运行的,Batch Duration需要很多资源,下一次Batch Duration就不需要那么多资源了,调整资源的时候还没调整完Batch Duration运行就已经过期了。这个时候调整时间间隔。

Spark Streaming资源动态申请

1. 在SparkContext中默认是不开启动态资源分配的,但是可以通过手动在SparkConf中配置。

// Optionally scale number of executors dynamically based on workload. Exposed for testing.val dynamicAllocationEnabled = Utils.isDynamicAllocationEnabled(_conf)if (!dynamicAllocationEnabled && //参数配置是否开启资源动态分配_conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
  logWarning("Dynamic Allocation and num executors both set, thus dynamic allocation disabled.")
}

_executorAllocationManager =
  if (dynamicAllocationEnabled) {    Some(new ExecutorAllocationManager(this, listenerBus, _conf))
  } else {    None
  }
_executorAllocationManager.foreach(_.start())
  1. ExecutorAllocationManager: 有定时器会不断的去扫描Executor的情况,正在运行的Stage,要运行在不同的Executor中,要么增加Executor或者减少。
  2. ExecutorAllocationManager中schedule方法会被周期性触发进行资源动态调整。
/** * This is called at a fixed interval to regulate the number of pending executor requests * and number of executors running. * * First, adjust our requested executors based on the add time and our current needs. * Then, if the remove time for an existing executor has expired, kill the executor. * * This is factored out into its own method for testing. */private def schedule(): Unit = synchronized {  val now = clock.getTimeMillis

  updateAndSyncNumExecutorsTarget(now)

  removeTimes.retain { case (executorId, expireTime) =>
    val expired = now >= expireTime    if (expired) {
      initializing = false
      removeExecutor(executorId)
    }    !expired
  }
}
  1. 在ExecutorAllocationManager中会在线程池中定时器会不断的运行schedule.
/** * Register for scheduler callbacks to decide when to add and remove executors, and start * the scheduling task. */def start(): Unit = {
  listenerBus.addListener(listener)  val scheduleTask = new Runnable() {    override def run(): Unit = {      try {
        schedule()
      } catch {        case ct: ControlThrowable =>
          throw ct        case t: Throwable =>
          logWarning(s"Uncaught exception in thread ${Thread.currentThread().getName}", t)
      }
    }
  }// intervalMillis定时器触发时间
  executor.scheduleAtFixedRate(scheduleTask, 0, intervalMillis, TimeUnit.MILLISECONDS)
}

动态控制消费速率: Spark Streaming提供了一种弹性机制,流进来的速度和处理速度的关系,是否来得及处理数据。如果不能来得及的话,他会自动动态控制数据流进来的速度,spark.streaming.backpressure.enabled参数设置。

动态控制消费速率的原理可参考论文 Adaptive Stream Processing using Dynamic Batch Sizing

备注:

1、DT大数据梦工厂微信公众号DT_Spark 
2、IMF晚8点大数据实战YY直播频道号:68917580
3、新浪微博: http://www.weibo.com/ilovepains

时间: 05-30

第17课:Spark Streaming资源动态申请和动态控制消费速率原理剖析的相关文章

(版本定制)第17课:Spark Streaming资源动态申请和动态控制消费速率原理剖析

本期内容: 1.Spark Streaming资源动态分配 2.Spark Streaming动态控制消费速率 为什么需要动态? a)Spark默认情况下粗粒度的,先分配好资源再计算.对于Spark Streaming而言有高峰值和低峰值,但是他们需要的资源是不一样的,如果按照高峰值的角度的话,就会有大量的资源浪费. b) Spark Streaming不断的运行,对资源消耗和管理也是我们要考虑的因素. Spark Streaming资源动态调整的时候会面临挑战: Spark Streaming

Spark Streaming资源动态申请和动态控制消费速率剖析

本期内容 : Spark Streaming资源动态分配 Spark Streaming动态控制消费速率 为什么需要动态处理 : Spark 属于粗粒度资源分配,也就是在默认情况下是先分配好资源然后再进行计算,粗粒度有个好处,因为资源是提前给你分配好,当有计算任务的时候直接使用就可以了, 粗粒度不好的方面就是从Spark  Streaming角度讲有高峰值.低峰值,在高与低峰值时候需要的资源是不一样的,如果资源分配按照高峰值考虑的话,在低峰值就是对资源的浪费, 随着Spark Streaming

Spark Streaming发行版笔记17:资源动态分配和动态控制消费速率

本篇从二个方面讲解: 高级特性: 1.Spark Streaming资源动态分配 2.Spark Streaming动态控制消费速率 原理剖析,动态控制消费速率其后面存在一套理论,资源动态分配也有一套理论. 先讲理论,后面讨论. 为什么要动态资源分配和动态控制速率? Spark默认是先分配资源,然后计算:粗粒度的分配方式,资源提前分配好,有计算任务提前分配好资源: 不好的地方:从Spark Streaming角度讲有高峰值和低峰值,如果资源分配从高峰值.低峰值考虑都有大量资源的浪费. 其实当年S

Spark Streaming中动态Batch Size深入及RateController解析

本期内容 : BatchDuration与 Process Time 动态Batch Size Spark Streaming中有很多算子,是否每一个算子都是预期中的类似线性规律的时间消耗呢? 例如:join操作和普通Map操作的处理数据的时间消耗是否会呈现出一致的线性规律呢,也就是说,并非数据量规模越大就是简单加大BatchDuration 就可以解决问题的,数据量是一个方面,计算的算子也是一个考量的因素. 使用BatchSize来适配我们的流处理程序 : 线上的处理程序越来越重要,流入的数据

第82课 Spark Streaming第一课 案例动手实战并在电光石火间理解其工作原理

本课内容提要: (1)什么是流处理以及Spark Streaming主要介绍 (2)Spark Streaming初体验 一.什么是流处理以及Spark Streaming主要介绍 流(Streaming),在大数据时代为数据流处理,就像水流一样,是数据流:既然是数据流处理,就会想到数据的流入.数据的加工.数据的流出. 日常工作.生活中数据来源很多不同的地方.例如:工业时代的汽车制造.监控设备.工业设备会产生很多源数据:信息时代的电商网站.日志服务器.社交网络.金融交易系统.黑客攻击.垃圾邮件.

第90讲,Spark streaming基于kafka 以Receiver方式获取数据 原理和案例实战

1:SparkSteaming基于kafka获取数据的方式,主要有俩种,即Receiver和Derict,基于Receiver的方式,是sparkStreaming给我们提供了kafka访问的高层api的封装,而基于Direct的方式,就是直接访问,在sparkSteaming中直接去操作kafka中的数据,不需要前面的高层api的封装.而Direct的方式,可以对kafka进行更好的控制!同时性能也更好. 2:实际上做kafka receiver的时候,通过receiver来获取数据,这个时候

Alex 的 Hadoop 菜鸟教程: 第17课 Spark 安装以及使用教程

声明 本文基于Centos6.x + CDH 5.x 本文基于CSDN的markdown编辑器写成,csdn终于支持markdown了,高兴! Spark是什么 Spark是Apache的顶级项目.项目背景是 Hadoop 的 MapReduce 太挫太慢了,于是有人就做了Spark,目前Spark声称在内存中比Hadoop快100倍,在磁盘上比Hadoop快10倍. 安装Spark spark有5个组件 spark-core: spark核心包 spark-worker: spark-work

Spark Streaming性能优化: 如何在生产环境下应对流数据峰值巨变

1.为什么引入Backpressure 默认情况下,Spark Streaming通过Receiver以生产者生产数据的速率接收数据,计算过程中会出现batch processing time > batch interval的情况,其中batch processing time 为实际计算一个批次花费时间, batch interval为Streaming应用设置的批处理间隔.这意味着Spark Streaming的数据接收速率高于Spark从队列中移除数据的速率,也就是数据处理能力低,在设置

Spark Streaming编程指南

本文基于Spark Streaming Programming Guide原文翻译, 加上一些自己的理解和小实验的结果. 一.概述 Spark Streaming是基于Core Spark API的可扩展,高吞吐量,并具有容错能力的用于处理实时数据流的一个组件.Spark Streaming可以接收各种数据源传递来的数据,比如Kafka, Flume, Kinesis或者TCP等,对接收到的数据还可以使用一些用高阶函数(比如map, reduce, join及window)进行封装的复杂算法做进