前言
前面,有分享过基于receiver的,实际上,看到receiver based Dstream大家就对阅读提不起兴趣了,实际上这是错误的,基于receiver的才是spark streaming根本,虽然direct stream才更合适。但是,我们从基于receiver可以学到很多内容,最重要的spark streaming实现原理,数据本地性等。详细可以再去阅读一下:
direct dstream运行架构图
对比
对比receiver based的Dstream和direct Dstream
a无需启动receiver,减少不必要的cpu占用
b减少了receiver接收数据,写入blockmanager,然后运行时再通过blockid,网络传输,磁盘读区,来获取数据这个过程。提升了效率。
c无需wal,进一步减少磁盘读写。
d可以通过手动维护offset来实现精确的一次消费。
eDstream中生成的RDD,并不是blockrdd,而是kafkardd,kafkardd是和kafka分区一一对应的,更便于我们把控并行度。
f数据本地性的问题,导致receiver存在的机器会运行过多的任务,会导致有些executor空闲。
而kafkardd,在compute函数里,会使用simpleconsumer,根据指定的topic,分区,offset范围,去kafka读取数据。010版本以后,又存在假如kafka和spark运行于同一集群,会有数据本性的概念。
数据本地性
spark streaming与kafka 082结合生成的rdd,数据本地性计算方式如下:
override defgetPreferredLocations(thePart: Partition):Seq[String] = {
valpart = thePart.asInstanceOf[KafkaRDDPartition]
// TODO is additional hostname resolution necessary here
Seq(part.host)
}
spark streaming 与kafka 010结合生成的rdd,数据本地性计算方式如下:
override defgetPreferredLocations(thePart: Partition):Seq[String] = {
// The intention is best-effort consistent executor for a given topicpartition,
// so that caching consumers can be effective.
// TODO what about hosts specified by ip vs name
valpart = thePart.asInstanceOf[KafkaRDDPartition]
valallExecs = executors()
valtp = part.topicPartition
valprefHost = preferredHosts.get(tp)
valprefExecs =if(null== prefHost) allExecselseallExecs.filter(_.host == prefHost)
valexecs =if(prefExecs.isEmpty) allExecselseprefExecs
if(execs.isEmpty) {
Seq.empty
}else{
// execs is sorted, tp.hashCode depends only on topic and partition, so consistent index
valindex = Math.floorMod(tp.hashCode,execs.length)
valchosen = execs(index)
Seq(chosen.toString)
}
}
对于 与kafka010结合的注意事项,实际上以前浪尖也翻译过一篇文章。
限速
限速,很多人使用姿势不对,详细的原理可以参看
具体配置参数详解,可以参考:
详细的视频讲解,可以观看:
更多资料,可以加入知识星球获取
领取专属 10元无门槛券
私享最新 技术干货