详解direct Dstream

前言

前面，有分享过基于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结合的注意事项，实际上以前浪尖也翻译过一篇文章。

限速

限速，很多人使用姿势不对，详细的原理可以参看

具体配置参数详解，可以参考：

详细的视频讲解，可以观看：

更多资料，可以加入知识星球获取