SparkStreaming获取kafka数据的两种方式：Receiver与Direct的方式

时间 2019-11-05

标签 sparkstreaming 获取 kafka 数据两种方式 receiver direct 栏目 Kafka 繁體版

原文原文链接

目录html

Receiverjava
Directapache
Direct代码bootstrap

简单理解为：Receiver方式是经过zookeeper来链接kafka队列，Direct方式是直接链接到kafka的节点上获取数据api

Receiver

使用Kafka的高层次Consumer API来实现。receiver从Kafka中获取的数据都存储在Spark Executor的内存中，而后Spark Streaming启动的job会去处理那些数据。然而，在默认的配置下，这种方式可能会由于底层的失败而丢失数据。若是要启用高可靠机制，让数据零丢失，就必须启用Spark Streaming的预写日志机制（Write Ahead Log，WAL）。该机制会同步地将接收到的Kafka数据写入分布式文件系统（好比HDFS）上的预写日志中。因此，即便底层节点出现了失败，也可使用预写日志中的数据进行恢复。分布式

注意事项：ide

一、Kafka中topic的partition与Spark中RDD的partition是没有关系的，所以，在KafkaUtils.createStream()中，提升partition的数量，只会增长Receiver的数量，也就是读取Kafka中topic partition的线程数量，不会增长Spark处理数据的并行度。性能

二、能够建立多个Kafka输入DStream，使用不一样的consumer group和topic，来经过多个receiver并行接收数据。spa

三、若是基于容错的文件系统，好比HDFS，启用了预写日志机制，接收到的数据都会被复制一份到预写日志中。所以，在KafkaUtils.createStream()中，设置的持久化级别是StorageLevel.MEMORY_AND_DISK_SER。线程

回到顶部

Direct

Spark1.3中引入Direct方式，用来替代掉使用Receiver接收数据，这种方式会周期性地查询Kafka，得到每一个topic+partition的最新的offset，从而定义每一个batch的offset的范围。当处理数据的job启动时，就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。

这种方式有以下优势：

一、简化并行读取：若是要读取多个partition，不须要建立多个输入DStream，而后对它们进行union操做。Spark会建立跟Kafka partition同样多的RDD partition，而且会并行从Kafka中读取数据。因此在Kafka partition和RDD partition之间，有一个一对一的映射关系。

二、高性能：若是要保证零数据丢失，在基于receiver的方式中，须要开启WAL机制。这种方式其实效率低下，由于数据实际上被复制了两份，Kafka本身自己就有高可靠的机制会对数据复制一份，而这里又会复制一份到WAL中。而基于direct的方式，不依赖Receiver，不须要开启WAL机制，只要Kafka中做了数据的复制，那么就能够经过Kafka的副本进行恢复。

三、一次且仅一次的事务机制：基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制能够保证数据零丢失的高可靠性，可是却没法保证数据被处理一次且仅一次，可能会处理两次。由于Spark和ZooKeeper之间多是不一样步的。基于direct的方式，使用kafka的简单api，Spark Streaming本身就负责追踪消费的offset，并保存在checkpoint中。Spark本身必定是同步的，所以能够保证数据是消费一次且仅消费一次。因为数据消费偏移量是保存在checkpoint中，所以，若是后续想使用kafka高级API消费数据，须要手动的更新zookeeper中的偏移量

回到顶部

Direct代码

 
      package 
      bigdata.spark

 
      import 
      kafka.serializer.{StringDecoder, Decoder}

 
      import 
      org.apache.spark.streaming.kafka.KafkaUtils

 
      import 
      org.apache.spark.streaming.{Seconds, StreamingContext}

 
      import 
      org.apache.spark.{SparkContext, SparkConf}

 
      import 
      scala.reflect.ClassTag

/**

 
      * Created by Administrator on 2017/4/28.

*/

 
      object SparkStreamDemo { 
     

 
      def main(args: Array[String]) {

 
      val conf =  
      new 
      SparkConf()

 
      conf.setAppName( 
      "spark_streaming" 
      )

 
      conf.setMaster( 
      "local[*]" 
      )

 
      val sc =  
      new 
      SparkContext(conf)

 
      sc.setCheckpointDir( 
      "D:/checkpoints" 
      )

 
      sc.setLogLevel( 
      "ERROR" 
      )

 
           
      val ssc =  
      new 
      StreamingContext(sc, Seconds( 
      5 
      )) 
     

 
      // val topics = Map("spark" -> 2)

 
      val kafkaParams = Map[String, String](

 
      "bootstrap.servers" 
      ->  
      "m1:9092,m2:9092,m3:9092" 
      ,

 
      "group.id" 
      ->  
      "spark" 
      ,

 
      "auto.offset.reset" 
      ->  
      "smallest"

)

 
      // 直连方式拉取数据，这种方式不会修改数据的偏移量，须要手动的更新

 
      val lines =  KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set( 
      "spark" 
      )).map(_._2)

 
      // val lines = KafkaUtils.createStream(ssc, "m1:2181,m2:2181,m3:2181", "spark", topics).map(_._2)

 
           
      val ds1 = lines.flatMap(_.split( 
      " " 
      )).map((_,  
      1 
      )) 
     

 
      val ds2 = ds1.updateStateByKey[Int]((x:Seq[Int], y:Option[Int]) => {

 
      Some(x.sum + y.getOrElse( 
      0 
      ))

})

 
      ds2.print()

 
      ssc.start()

 
      ssc.awaitTermination()

}

}

　转自：http://www.javashuo.com/article/p-smzapiqr-cy.html