Spark学习（7）- Spark Streaming整合Kafka

桃扇骨 2022-03-10 03:42 228阅读 0赞

**流程：**

1.  创建maven项目
2.  引入相应的依赖
3.  编写代码
4.  发布

**第一种方式：KafkaUtils.createDstream方式（基于kafka高级Api-----偏移量由zk保存）**

import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.kafka.KafkaUtils
    
    import scala.collection.immutable
    
    //todo:sparkStreaming整合kafka---基于receiver（高级api）
    object SparkStreamingKafkaReceiver {
      def main(args: Array[String]): Unit = {
          //1、创建sparkConf
          val sparkConf: SparkConf = new SparkConf()
                                      .setAppName("SparkStreamingKafkaReceiver")
                                      .setMaster("local[4]") //线程数要大于receiver个数
                                        .set("spark.streaming.receiver.writeAheadLog.enable","true")
                                          //表示开启WAL预写日志，保证数据源的可靠性
          //2、创建sparkContext
          val sc = new SparkContext(sparkConf)
          sc.setLogLevel("WARN")
          //3、创建streamingContext
          val ssc = new StreamingContext(sc,Seconds(5))
            ssc.checkpoint("./spark_receiver")
          //4、准备zk地址
          val zkQuorum="node1:2181,node2:2181,node3:2181"
          //5、准备groupId
          val groupId="spark_receiver"
          //6、定义topic   当前这个value并不是topic对应的分区数，而是针对于每一个分区使用多少个线程去消费
         val topics=Map("spark_01" ->2)
          //7、KafkaUtils.createStream 去接受kafka中topic数据
          //(String, String) 前面一个string表示topic名称，后面一个string表示topic中的数据
        //使用多个reveiver接受器去接受kafka中topic数据
        val dstreamSeq: immutable.IndexedSeq[ReceiverInputDStream[(String, String)]] = (1 to 3).map(x => {
          val stream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)
          stream
          }
        )
    
        //利用streamingcontext调用union,获取得到所有receiver接受器的数据
        val totalDstream: DStream[(String, String)] = ssc.union(dstreamSeq)
    
         //8、获取kafka中topic的数据
        val topicData: DStream[String] = totalDstream.map(_._2)
         //9、切分每一行
          val wordAndOne: DStream[(String, Int)] = topicData.flatMap(_.split(" ")).map((_,1))
         //10、相同单词出现的次数累加
          val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
        //11、打印结果数据
          result.print()
    
    
        //12、开启流式计算
          ssc.start()
          ssc.awaitTermination()
    
      }
    }

**第二种方式：KafkaUtils.createDirectStream方式（基于kafka低级Api-----偏移量由客户端程序保存）**

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:sparkSteaming整合kafka----采用direct(低级Api)
    object SparkStreamingKafkaDirect {
      def main(args: Array[String]): Unit = {
          //1、创建sparkConf
          val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingKafkaDirect").setMaster("local[2]")
          //2、创建sparkContext
          val sc = new SparkContext(sparkConf)
          sc.setLogLevel("WARN")
          //3、创建streamingcontext
          val ssc = new StreamingContext(sc,Seconds(5))
          ssc.checkpoint("./spark_direct") //它会保存topic的偏移量
          //4、准备kafka参数
          val kafkaParams=Map("metadata.broker.list"->"node1:9092,node2:9092,node3:9092","group.id" ->"spark_direct")
          //5、准备topic
          val topics=Set("spark_01")
          //6、获取kafka中的数据
          val dstream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)
          //7、获取topic中的数据
            val data: DStream[String] = dstream.map(_._2)
          //8、切分每一行，每个单词计为1，把相同单词出现的次数累加
            val result: DStream[(String, Int)] = data.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
    
          //9、打印结果数据
            result.print()
    
          //10、开启流式计算
            ssc.start()
            ssc.awaitTermination()
    
      }
    }