阅读最新的spark kafka流媒体

Question

我希望只使用kafka读取火花流中的最新消息,但它也会获取过去的数据

如何在KafkaUtil中为spark设置auto.offset.reset

JavaPairReceiverInputDStream<String, String> messages =
            KafkaUtils.createStream(jssc, args[0], args[1], topicMap);

如何将conf设置为仅获取当前消息.请举个例子.

在此先感谢,还有另一个线程

但还不够,请帮助我.提前致谢.

Answer 1

您需要在KafkaUtils对象中使用此方法:

 def createStream[K, V, U <: Decoder[_], T <: Decoder[_]](
      jssc: JavaStreamingContext,
      keyTypeClass: Class[K],
      valueTypeClass: Class[V],
      keyDecoderClass: Class[U],
      valueDecoderClass: Class[T],
      kafkaParams: JMap[String, String],
      topics: JMap[String, JInt],
      storageLevel: StorageLevel
    )

根据Spark版本,您不能使用java.有一个错误.

如果您使用的是Spark 1.1.0,则需要在kafkaParams参数中添加以下属性:

"auto.offset.reset","最大"

另一种解决方法是随机生成groupId前缀,但这很糟糕.