spark小案例——sparkstreaming消费Kafka

Posted 2022-12-09 Z-hhhhh

使用sparkstreaming消费Kafka的数据，实现word count

依赖

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
    <version>3.0.0</version>
</dependency>
<dependency>
     <groupId>com.fasterxml.jackson.core</groupId>
     <artifactId>jackson-core</artifactId>
     <version>2.10.1</version>
</dependency>

实现wordcount代码

val conf = new SparkConf().setAppName("StreamWordCount").setMaster("local[*]")
    val sc = new StreamingContext(conf, Seconds(3))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop01:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "group1",
    /**
     * earliest
     * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
     * latest
     * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
     * none
     * topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
     */
      "auto.offset.reset" -> "earliest"
    )

    /**
     * LocationStrategies.PreferBrokers() 仅仅在你 spark 的 executor 在相同的节点上，优先分配到存在  kafka broker 的机器上；
     * LocationStrategies.PreferConsistent(); 大多数情况下使用，一致性的方式分配分区所有 executor 上。（主要是为了分布均匀）
     * 新的Kafka使用者API将预先获取消息到缓冲区。因此，出于性能原因，Spark集成将缓存的消费者保留在执行程序上（而不是为每个批处理重新创建它们），并且更喜欢在具有适当使用者的主机位置上安排分区，这一点很重要。
     * 在大多数情况下，您应该使用LocationStrategies.PreferConsistent，如上所示。这将在可用执行程序之间均匀分配分区。如果您的执行程序与Kafka代理在同一主机上，请使用PreferBrokers，它更愿意为该分区安排Kafka领导者的分区。
     */
    val topics = Array("topic01")
    val stream = KafkaUtils.createDirectStream[String, String](
      sc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )


    /*  val kafkaStream = stream.map(record => (record.key, record.value))
        val words = kafkaStream.map(_._2)
        val pairs = words.map  x => (x,1) 
        val wordCounts = pairs.reduceByKey(_+_)
    */
    val value: DStream[String] = stream.map(record => record.value())
    value.flatMap(_.split(" "))
        .map((_,1))
        .reduceByKey(_+_)
        .print()

    sc.start()
    sc.awaitTermination()

启动Kafka,创建Kafka producer

kafka-console-producer.sh --broker-list hadoop01:9092 --to topic01