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

阳光穿透心脏的1/2处 2023-05-29 03:38 7阅读 0赞

# 前言 #

*  kafka作为一个实时的分布式消息队列，实时的生产和消费消息，这里我们可以利用**SparkStreaming实时地读取kafka中的数据**，然后进行相关计算。
 *  在Spark1.3版本后，**KafkaUtils里面提供了两个创建dstream的方法，一种为KafkaUtils.createDstream，另一种为KafkaUtils.createDirectStream**。

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

**此种方式现在已被淘汰,大家可作为了解学习~**

*  KafkaUtils.createDstream(ssc, \[zk\], \[group id\], \[per-topic,partitions\] ) 使用了**receivers**接收器来接收数据，利用的是**Kafka高层次的消费者api**.
 *  对于所有的**receivers接收到的数据将会保存在Spark executors**中.
 *  然后通过Spark Streaming启动job来处理这些数据，默认会丢失.
 *  可启用WAL日志，它同步将接受到数据保存到分布式文件系统上比如HDFS。 所以数据在出错的情况下可以恢复出来 。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]
 *  Spark集群中的某个executor中有一个**receiver线程**, 这个线程负责从**kafka中获取数据**.  
    ![在这里插入图片描述][2019110316035359.png]
 *  当**receiver线程**接收到数据后会做备份处理,即把数据备份到其他的executor中,也可能会备份到这个receiver线程所以在节点的executor中.
 *  当备份完毕后该现场会把每个**partition**的**消费偏移量**在**zookeeper**中修改,(新版本的kafka的offset保存在kafka集群中)
 *  修改完**offset**后,该receiver线程会把**消费的数据**告诉**Driver**.
 *  **Driver**分发任务时会根据每个**executor**上的数据,根据数据本地性发送.

# 2. 代码实现 #

*  pom文件中添加kafka依赖

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

*  编写Spark Streaming程序

package cn.acece.sparkStreamingKafkaTest
    
    import org.apache.spark.streaming.dstream.{ DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.{ SparkConf, SparkContext}
    /** * 从kafka中拉取数据 * 读取数据时, consumer记录的offset发送回kafka中, 保存在zk中 * 需要开始WAL日志保存模式防止数据丢失, 需要设置检查点 */
    object SparkStreamingPollKafkaHighAPI { 
      def main(args: Array[String]): Unit = { 
        //1. 初始化参数,conf, sc, ssc
        val conf: SparkConf = new SparkConf()
          .setAppName("SparkStreamingPollKafkaHighAPI")
          .setMaster("local[2]")
          .set("spark.streaming.receiver.writeAheadLog.enable", "true")
        val sc: SparkContext = new SparkContext(conf)
        sc.setLogLevel("WARN")
    
        val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))
        //设置检查点, 开启WLA日志保存机制就要设置检查点
        ssc.checkpoint("./check")
    
        //2. 从kafka中拉取数据, KafKaUtil
        val zkQuorum = "node01:2181,node02:2181,node03:2181"
        val groupId = "group"
        //这里的1, 代表每一个分区被N个消费者消费
        val topics = Map[String,Int]("itcast"->1)
        val receiver: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc,zkQuorum,groupId,topics)
    
        //3. 从主体中获取具体的数据, 也就是value值, key是offect
        val lines: DStream[String] = receiver.map(_._2)
    
        //4. 单词计数
        val words: DStream[String] = lines.flatMap(_.split(" "))
        val wordAnd1: DStream[(String, Int)] = words.map((_,1))
        val result: DStream[(String, Int)] = wordAnd1.reduceByKeyAndWindow((x:Int,y:Int) => x+y, Seconds(5),Seconds(5))
    
        //5. 打印
        result.print()
    
        //6. 开启流模式
        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

![在这里插入图片描述][20191103164239951.png]

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

# 4. 总结 #

*  通过这种方式实现，刚开始的时候系统正常运行，没有发现问题，但是如果系统异常重新启动sparkstreaming程序后，**发现程序会重复处理已经处理过的数据**.
 *  这种基于receiver的方式，是使用**Kafka的高级API**，**topic的offset偏移量在ZooKeeper中**。
 *  这是消费Kafka数据的传统方式。这种方式**配合着WAL机制可以保证数据零丢失的高可靠性，但是却无法保证数据只被处理一次，可能会处理两次**。
 *  因为**Spark和ZooKeeper之间可能是不同步的**。官方现在也已经不推荐这种整合方式，我们使用官网推荐的第二种方式kafkaUtils的createDirectStream()方式。

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