SparkStreaming---wordcount(kafka)

Posted 2021-06-29 Shall潇

本文主要讲：利用 SparkStreaming 方式读取并处理 kafka中的数据，最后存储到 kafka

文章目录

• 一、导入依赖
• 二、编写代码
• 三、测试

一、导入依赖

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.11</artifactId>
  <version>2.4.7</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.11</artifactId>
  <version>2.4.7</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming_2.11</artifactId>
  <version>2.4.7</version>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
  <version>2.4.7</version>
</dependency>

二、编写代码

package kafkademo

import java.util

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

/*
* @Description: 统计WordCount,将kafka中的数据读取出来，处理好后并存入kafka
*
* DStream ---> 一堆RDD
* DStream.foreachRDD 遍历所有的RDD，对每个RDD进行操作
* RDD.foreachPartition 处理的是 record 对象，record.value获取 kafka 的value。 采用foreachPartition是减少task内存压力
* */

object SparkStreamKafkaSourceToKafkaSinkWC {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("kafkaSourceKafkaSink")
    val streamingContext = new StreamingContext(conf,Seconds(5))

    // 设置检查点
    streamingContext.checkpoint("checkpoint")

    // 设置kafka配置信息
    val kafkaParams = Map(
      (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "192.168.XXX.100:9092"),
      (ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG->"org.apache.kafka.common.serialization.StringDeserializer"),
      (ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG->"org.apache.kafka.common.serialization.StringDeserializer"),
      (ConsumerConfig.GROUP_ID_CONFIG->"kafkaGroup1")
    )

    val kafkaStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      streamingContext,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe(Set("sparkKafka"), kafkaParams)
    )

    val mapStream: DStream[(String, Int)] = kafkaStream.flatMap(x=>x.value().split("\\\\s+")).map(x=>(x,1))
    val wcDS: DStream[(String, Int)] = mapStream.reduceByKey(_+_)

    wcDS.foreachRDD(rdd=>{
      rdd.foreachPartition(records=>{
        val prop = new util.HashMap[String,Object]()
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.XXX.100:9092")
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")

        val producer = new KafkaProducer[String,String](prop)
        records.foreach(record=>{
          val re = new ProducerRecord[String,String]("sparkKafkaOut","",record._1+":"+record._2)

          producer.send(re)
        })
      })
    })

    streamingContext.start()
    streamingContext.awaitTermination()
  }
}

三、测试

创建并开启 sparkKafka 的生产者，并生产数据
从 sparkKafkaOut 消费数据