sparkstreaming监听hdfs目录如何终止_Spark笔试题：Spark Streaming 反压机制

Spark Streaming 反压机制是1.5版本推出的特性，用来解决处理速度比摄入速度慢的情况，简单来讲就是做流量控制。当批处理时间(Batch Processing Time)大于批次间隔(Batch Interval，即BatchDuration)时,说明处理数据的速度小于数据摄入的速度，持续时间过⻓或源头数据暴增，容易造成 数据在内存中堆积，最终导致Executor OOM。反压就是来解决这个问题的。
spark streaming的消费数据源方式有两种：
1. 若是基于Receiver的数据源，可以通过设置spark.streaming.receiver.maxRate来控制最大输入 速率；
2. 若是基于Direct的数据源(如Kafka Direct Stream)，则可以通过设置spark.streaming.kafka.maxRatePerPartition来控制最大输入速率。

当然，在事先经过压测，且流量高峰不会超过预期的情况下，设置这些参数一般没什么问题。但最大值不代表是最优值，最好还能根据每个批次处理情况来动态预估下个批次最优速率。

在Spark 1.5.0以上，就可通过背压机制来实现。开启反压机制，即设置spark.streaming.backpressure.enabled为true，Spark Streaming会自动根据处理能力来调整输入速率，从而在流量高峰时仍能保证最大的吞吐和性能

Spark Streaming的反压机制中，有以下几个重要的组件：

第一:RateController
RateController 组件是 JobScheduler 的监听器，主要监听集群所有作业的提交、运行、完成情况，并从 BatchInfo 实例中获取以下信息，交给速率估算器(RateEstimator)做速率的估算。

• 当前批次任务处理完成的时间戳 (processingEndTime)
• 该批次从第一个 job 到最后一个 job 的实际处理时⻓ (processingDelay)
• 该批次的调度时延，即从被提交到 JobScheduler 到第一个 job 开始处理的时⻓
  (schedulingDelay)
• 该批次输入数据的总条数(numRecords)

第二:RateEstimator
Spark 2.x 只支持基于 PID 的速率估算器，这里只讨论这种实现。基于 PID 的速率估算器简单地说就是它把收集到的数据(当前批次速率)和一个设定值(上一批次速率)进行比较，然后用它们 之间的差计算新的输入值，估算出一个合适的用于下一批次的流量阈值。这里估算出来的值就是流 量的阈值，用于更新每秒能够处理的最大记录数

第三:RateLimiter

RateController和RateEstimator组件都是在Driver端用于更新最大速度的，而RateLimiter是用于接收到Driver的更新通知之后更新Executor的最大处理速率的组件。RateLimiter是一个抽象类，它并不是Spark本身实现 的，而是借助了第三方Google的GuavaRateLimiter来产生的。它实质上是一个限流器，也可以叫 做令牌，如果Executor中task每秒计算的速度大于该值则阻塞，如果小于该值则通过，将流数据加 入缓存中进行计算。

*反压机制真正起作用时需要至少处理一个批：由于反压机制需要根据当前批的速率，预估新批的速率，所以反压机制真正起作用前，应至少保证处理一个批。*如何保证反压机制真正起作用前应用不会崩溃：要保证反压机制真正起作用前应用不会崩溃,需要控制每个批次最大摄入速率。若为Direct Stream，如Kafka Direct Stream,则可以通过spark.streaming.kafka.maxRatePerPartition参数来控制。此参数代表了每秒每个分区最大摄入的数据条数。假设BatchDuration为10秒,spark.streaming.kafka.maxRatePerPartition为12条,kafka topic分区数为3个，则一个批(Batch)最大读取的数据条数为360条(3*12*10=360)。同时，需要注意，该参数也代表了整个应用生命周期中的最大速率，即使是背压调整的最大值也不会超过该参数。

反压相关的参数

总结

以上是生活随笔为你收集整理的sparkstreaming监听hdfs目录如何终止_Spark笔试题：Spark Streaming 反压机制的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。