Spark Streaming快速入门系列(3) | Kafka 数据源

谁借莪１个温暖的怀抱￠ 2022-12-03 15:48 140阅读 0赞

### 目录 ###

*  一.Kafka选型
 *  二.kafka和streaming理论指导
 *  三.从kafka消费数据1
 *  三.从kafka消费数据2
 *  三.从kafka消费数据3

# 一.Kafka选型 #

[http://spark.apache.org/docs/2.1.1/streaming-programming-guide.html][http_spark.apache.org_docs_2.1.1_streaming-programming-guide.html]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]  
[http://spark.apache.org/docs/2.1.1/streaming-kafka-integration.html][http_spark.apache.org_docs_2.1.1_streaming-kafka-integration.html]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]

# 二.kafka和streaming理论指导 #

[http://spark.apache.org/docs/2.1.1/streaming-kafka-0-8-integration.html][http_spark.apache.org_docs_2.1.1_streaming-kafka-0-8-integration.html]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]

# 三.从kafka消费数据1 #

*  1.用法及说明  
    在工程中需要引入 Maven 依赖 spark-streaming-kafka\_2.11来使用它。  
    包内提供的 KafkaUtils 对象可以在 StreamingContext和JavaStreamingContext中以你的 Kafka 消息创建出 DStream。  
    两个核心类：KafkaUtils、KafkaCluster
 *  2.导入依赖

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
        <version>2.1.1</version>
    </dependency>

*  3.代码实现

import kafka.serializer.StringDecoder
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object Kafka1 { 
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[2]").setAppName("WordCount2")
        val ssc = new StreamingContext(conf, Seconds(3))
    
        val params = Map[String, String](
          "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
          "group.id" -> "1602"
        )
    
        KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          params,
          Set("first1602")
        )
          .flatMap{ 
          case (_,v) =>
            v.split("\\W+")
        }
          .map((_,1))
            .reduceByKey(_+_)
            .print()
    
        ssc.start()
        ssc.awaitTermination()
      }
    
    }

*  4.测试kafka是否打通

bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first1602

![在这里插入图片描述][20200904114053540.png_pic_center]

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first1602

![在这里插入图片描述][2020090411415295.png_pic_center]

*  5.实现代码  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 4]  
缺点：数据会丢失

# 三.从kafka消费数据2 #

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
        <version>2.1.1</version>
    </dependency>

*  3.思考一下spark streaming关闭了 kafka发数据，再次打开spark streaming，怎样才能接受到数据  
    代码实现

import kafka.serializer.StringDecoder
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object Kafka2 { 
      def createSSC()={ 
        val conf = new SparkConf().setMaster("local[*]").setAppName("kafka2")
        val ssc = new StreamingContext(conf, Seconds(3))
        //把offset的跟踪在checkpoint中
        ssc.checkpoint("ck1")
    
        val params = Map[String, String](
          "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
          "group.id" -> "1602"
        )
    
        KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          params,
          Set("first1602")
        )
          .flatMap{ 
            case (_,v) =>
              v.split("\\W+")
          }
          .map((_,1))
          .reduceByKey(_+_)
          .print()
    
        ssc
      }
    
      def main(args: Array[String]): Unit = { 
        /* 从checkpoint中恢复一个streamingContext， 如果checkpoint不存在，则调用后面的函数去创建一个StreamingContext */
        val ssc = StreamingContext .getActiveOrCreate("ck1", createSSC)
    
        ssc.start()
        ssc.awaitTermination()
      }
    
    }

缺点：小文件太多  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 5]

# 三.从kafka消费数据3 #

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
        <version>2.1.1</version>
    </dependency>

*  3.从kafka消费数据2中产生大量小文件，怎么样解决？  
    解决方案：启动实时项目的实时读取偏移量+保存偏移量
 *  4.代码实现

import kafka.common.TopicAndPartition
    import kafka.message.MessageAndMetadata
    import kafka.serializer.StringDecoder
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.dstream.InputDStream
    import org.apache.spark.streaming.kafka.KafkaCluster.Err
    import org.apache.spark.streaming.kafka.{ HasOffsetRanges, KafkaCluster, KafkaUtils, OffsetRange}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object Kafka3 { 
      val groupId = "1602"
      val params = Map[String, String](
        "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
        "group.id" -> "1602")
      val topics = Set("first1602")
      //KafkaUtils KafkaCluster
      val cluster = new KafkaCluster(params)
    
      def readOffsets()={ 
        //最终返回的map
        var resultMap = Map[TopicAndPartition, Long]()
        val topicAndPartitionSetEither: Either[Err, Set[TopicAndPartition]] = cluster.getPartitions(topics)
        topicAndPartitionSetEither match { 
            //2.获取topic和分区的信息
          case Right(topicAndPartitionSet) =>
            //3.获取到分区信息和其他的offset
            val topicAndPartitionToLongEither: Either[Err, Map[TopicAndPartition, Long]] = cluster.getConsumerOffsets(groupId, topicAndPartitionSet)
            topicAndPartitionToLongEither match { 
                //没有每个topic的每个分区都已经存储过偏移量，表示曾经消费过，而且也维护过偏移量
              case Right(map)=>
                resultMap ++= map
    
                //表示这个topic的这个分区是第一次消费
              case  _ =>
                 topicAndPartitionSet.foreach(topicAndPartition => { 
                   resultMap += topicAndPartition -> 1L
                 })
            }
          case _ => //表示不存在任何topic
    
        }
        resultMap
      }
    
      def saveOffsets(stream: InputDStream[String]) = { 
        //保存offset一定从kafka消费到的直接那个stream保存
        //每个批次执行一次传递过去的函数
        stream.foreachRDD(rdd =>{ 
          var map = Map[TopicAndPartition,Long]()
          //如果这个rdd是来自kafka，可以强转成HasOffsetRanges
          //这类型就包含了，这次消费的offsets的信息
          val hasOffsetRanges: HasOffsetRanges = rdd.asInstanceOf[HasOffsetRanges]
          //如果有两个topic 三个分区 那么就有6个OffsetRange
          //所有分区的偏移量
          val ranges: Array[OffsetRange] = hasOffsetRanges.offsetRanges
          ranges.foreach(OffsetRange=>{ 
            val key = OffsetRange.topicAndPartition()
            val value = OffsetRange.untilOffset
            map += key -> value
          })
          cluster.setConsumerOffsets(groupId,map)
        })
      }
    
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[*]").setAppName("kafka2")
        val ssc = new StreamingContext(conf, Seconds(3))
    
        val sourceStream: InputDStream[String] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, String](
          ssc,
          params,
          readOffsets(),
          (handler: MessageAndMetadata[String, String]) => handler.message() // 从kafka读到数据的value
    
        )
        sourceStream
            .flatMap(_.split(("\\W+")))
            .map((_,1))
            .reduceByKey(_+_)
            .print(1000)
    
        saveOffsets(sourceStream)
    
        ssc.start()
        ssc.awaitTermination()
      }
    
    }

[http_spark.apache.org_docs_2.1.1_streaming-programming-guide.html]: http://spark.apache.org/docs/2.1.1/streaming-programming-guide.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/5cab280e38e34892b0387bbcb7646af1.png
[http_spark.apache.org_docs_2.1.1_streaming-kafka-integration.html]: http://spark.apache.org/docs/2.1.1/streaming-kafka-integration.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/c8ad7ef3be534ea08744620ae17d9058.png
[http_spark.apache.org_docs_2.1.1_streaming-kafka-0-8-integration.html]: http://spark.apache.org/docs/2.1.1/streaming-kafka-0-8-integration.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/475b801c68d749ed8a3066f75a44651d.png
[20200904114053540.png_pic_center]: /images/20221123/938e043786ea4b7e9a0f006e710acc0a.png
[2020090411415295.png_pic_center]: /images/20221123/91b420a8c9cb4cc2b15913289bbd2403.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/17dacb0a14ed4beb97dc5fe6efb8c6a2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/deaa6205c0a240e494fbfeca264a9e8e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/46d965c8fbc440849c618e08c3c93a79.png