使用KafkaUtils时手动将偏移范围提交给Kafka。createDirectStream

Question

我试图使用Kafka Utils Api从Kafka（0.10.0.0）到Spark（1.6.0）流媒体应用程序使用html" target="_blank">数据

Kafka提尔。createDirectStream[String，String，StringDecoder，StringDecoder]（ssc，kafkaParams，InputOpicSet）

要求是将偏移范围手动提交给Kafka本身。

请注意，当在java中使用Kafka消费者（或消费者）对象时，我们可以在参数中设置"enable.auto.commit"="false"后，使用委员会同步或委员会同步方法来实现这一点。

当使用KafkaUtils时，我无法找出同样的方法。

Answer 1

您可以传递"enable.auto.commit"="false"作为kafkaParams的一部分。事实上，您可以传递任何Kafka消费者设置作为其中的一部分。

    Map<String, Object> kafkaParams = new HashMap<>();
    kafkaParams.put("bootstrap.servers", "localhost:9092,anotherhost:9092");
    kafkaParams.put("key.deserializer", StringDeserializer.class);
    kafkaParams.put("value.deserializer", StringDeserializer.class);
    kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
    kafkaParams.put("enable.auto.commit", true); 
    //more kafka params goes here if needed
    JavaInputDStream<ConsumerRecord<String, String>> stream =
    KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, inputTopicsSet)

然后像这样手动提交偏移量，

stream.foreachRDD(rdd -> {
  OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges();

  // some time later, after outputs have completed
  ((CanCommitOffsets) stream.inputDStream()).commitAsync(offsetRanges);
});

参考：https://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html#creating-a-direct-stream

上述代码适用于spark 2.2.0，可能不适用于spark 1.6.0。