kafka只消费新消息-kafkatoconsumeonlynewmessages

There is two ways you can achieve this:

有两种方法可以实现这一目标:

1. Create a unique consumer group each time on restart and it will consume from the latest offset.

   每次重新启动时创建一个唯一的使用者组,它将使用最新的偏移量。

2. Use the direct approach instead of receiver based; here you have more control over how you consume but would have to update zookeeper manually to store your offsets. In the example below it will always start at latest offset.

   使用直接方法而不是基于接收器;在这里,您可以更好地控制使用方式,但必须手动更新zookeeper以存储偏移量。在下面的示例中,它始终以最新的偏移量开始。

   import org.apache.spark.streaming.kafka._
   val topicsSet = topics.split(",").toSet
   val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
   val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)
   

Documentation on direct approach here: https://spark.apache.org/docs/latest/streaming-kafka-integration.html

有关直接方法的文档:https://spark.apache.org/docs/latest/streaming-kafka-integration.html