DStream实战之Spark Streaming整合Kafka, 通过KafkaUtils.createDirectStream方式整合Kafka 38

深藏阁楼爱情的钟 2023-05-29 13:23 89阅读 0赞

# 前言 #

*  kafka作为一个实时的分布式消息队列，实时的生产和消费消息，这里我们可以利用SparkStreaming实时地读取kafka中的数据，然后进行相关计算。
 *  在Spark1.3版本后，KafkaUtils里面提供了两个创建dstream的方法，**一种为KafkaUtils.createDstream**(此种方法已被淘汰,博主上篇文章中已提过~)，另一种为**KafkaUtils.createDirectStream**(目前使用的也是这种方法)。

# 1. KafkaUtils.createDirectStream方式整合Kafka #

这种方式不同于Receiver接收数据，它**定期地从kafka的topic下对应的partition**中查询**最新的偏移量**，再根据**偏移量范围在每个batch里面处理数据**，Spark通过调用kafka简单的消费者Api（低级api）读取一定范围的数据。

*  消费的**offset**保存在**kafka集群中**(本台机器中),**不存在重复消费的情况**
 *  直接只用**executor**去获取数据,不存在丢失数据的情况.  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]
 *  首先Driver程序会**定时**(batchInterval)的向**executor**中发送任务(4个).

1.  **Driver怎么知道要把任务发送到哪个executor中呢?**  
    答: Driver会调用Kafka的接口获取某个partition位于哪个节点上,根据这个来获取信息并发送任务到指定的节点, 这就类似于Spark集群处理hdfs的文件数据, Spark是可以知道某些文件的block在哪些节点上, 就是Spark调用了hdfs的相关接口.
2.  为什么是4个任务  
    答: 这个个数由消费的topic的patition个数决定, 因为Spark会对每个partition开启一个任务, 所以任务数是kafka的某个topic的parition数.

*  当每个任务确定了处理的那个parition中的数据,则就有任务本身去kafka获取数据

**总结**:  
3. 目前公司中Spark Streaming整合Kafka采用此种方式的最多, 这样也有一个问题就是说当Kafka中某个topic加了ACK验证,那么这种方式是不能消费加了ACK的topic中的数据.  
4. 因为kafka客户端的ACK验证需要客户端配置一个环境变量在**System的Properties中**, 在local模式下可以实现. 因为local模式下启动一个虚拟机实例, 即只对应一个System. 而在集群模式下, 要启动多个进程, 即启动多个虚拟机实例, 所以System的全局属性没有办法配置.

# 2. 代码实现 #

*  添加pom依赖

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka_0-8_2.11</artifactId>
        <version>2.0.2</version>
    </dependency>

*  编写Spark Streaming程序  
    配置kafka集群的IP地址  
    使用**KafkaUtils.createDirectStream**生成Streaming流

package cn.acece.sparkStreamingKafkaTest
    
    import kafka.serializer.StringDecoder
    import org.apache.spark.{ SparkConf, SparkContext}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.streaming.dstream.{ DStream, InputDStream}
    import org.apache.spark.streaming.kafka.KafkaUtils
    
    //todo:利用sparkStreaming对接kafka实现单词计数----采用Direct(低级API)
    object SparkStreamingKafka_Direct { 
        def main(args: Array[String]): Unit = { 
          //1、创建sparkConf
          val sparkConf: SparkConf = new SparkConf()
            .setAppName("SparkStreamingKafka_Direct")
            .setMaster("local[2]")
          //2、创建sparkContext
          val sc = new SparkContext(sparkConf)
          sc.setLogLevel("WARN")
          //3、创建StreamingContext
          val ssc = new StreamingContext(sc,Seconds(5))
    ssc.checkpoint("./Kafka_Direct")
          //4、配置kafka相关参数
          val kafkaParams=Map("metadata.broker.list"->"node1:9092,node2:9092,node3:9092","group.id"->"Kafka_Direct")
          //5、定义topic
          val topics=Set("spark01")
          //6、通过 KafkaUtils.createDirectStream接受kafka数据，这里采用是kafka低级api偏移量不受zk管理
          val dstream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)
          //7、获取kafka中topic中的数据
            val topicData: DStream[String] = dstream.map(_._2)
          //8、切分每一行,每个单词计为1
          val wordAndOne: DStream[(String, Int)] = topicData.flatMap(_.split(" ")).map((_,1))
          //9、相同单词出现的次数累加
          val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
          //10、打印输出
          result.print()
          //开启计算
          ssc.start()
          ssc.awaitTermination()
      }
    }

# 3. 先启动Kafka再启动Spark Streaming程序,试试采集 #

*  启动zookeeper集群

zkServer.sh start

*  先启动kafka集群

kafka-server-start.sh /export/servers/kafka/config/server.properties

*  创建topic

kafka-topic.sh --create --zookeeper node01:2181 --replication-factor 1 --partitions 3 --topic kafka_spark

*  向topic中生产数据, 通过shell命令向topic发送消息

kafka-console-producer.sh --broker-list node01:9092 --topic kafka-spark

!\[\[在这里插入图片描述\](https://img-blog.csdnimg.cn/20191103164239951.png)

*  在IDEA中执行2中编写的Spark Streaming程序
 *  运行代码, 查看控制台结果  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191106162609187.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191106195916360.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ==,size_16,color_FFFFFF,t_70