how to specify consumer group in Kafka Spark Streaming using direct stream

6
votes

How to specify consumer group id for kafka spark streaming using direct stream API.

HashMap<String, String> kafkaParams = new HashMap<String, String>();
kafkaParams.put("metadata.broker.list", brokers);
kafkaParams.put("auto.offset.reset", "largest");
kafkaParams.put("group.id", "app1");

    JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
            jssc, 
            String.class, 
            String.class,
            StringDecoder.class, 
            StringDecoder.class, 
            kafkaParams, 
            topicsSet
    );

though i have specified the configuration not sure if missing something. using spark1.3

kafkaParams.put("group.id", "app1");
2
javaapache-sparkapache-kafkaspark-streamingkafka-consumer-api
It should be exactly as you're doing it. - Yuval Itzchakov
What do you mean by not sure if missing something? Please ask a specific question. Something like I tried X to achieve Y using library Z but got exception E with stacktrace S is appropriate from StackOverflow. - Debosmit Ray
@DebosmitRay I tried "group.id" using spark kafka direct stream to specify consumer group. Not getting any exception but want to know if this the right way to specify consumer group while using createDirectStream API method. Does it help now??? - Faisal Ahmed Siddiqui

2 Answers

7
votes

The direct stream API use the low level Kafka API, and as so doesn't use consumer groups in anyway. If you want to use consumer groups with Spark Streaming, you'll have to use the receiver based API.

Full details are available in the doc !

5
votes

createDirectStream in spark-streaming-kafka-0-8 does not support group mode, because it's using the low-level Kafka API.

But spark-streaming-kafka-0-10 supports group mode.

Consumer Configs

In 0.9.0.0 we introduced the new Java consumer as a replacement for the older Scala-based simple and high-level consumers. The configs for both new and old consumers are described below.

In the New Consumer Configs, it has the group.id item.

The Spark Streaming integration for Kafka 0.10 is using the new API. https://spark.apache.org/docs/2.1.1/streaming-kafka-0-10-integration.html

The Spark Streaming integration for Kafka 0.10 is similar in design to the 0.8 Direct Stream approach. It provides simple parallelism, 1:1 correspondence between Kafka partitions and Spark partitions, and access to offsets and metadata. However, because the newer integration uses the new Kafka consumer API instead of the simple API, there are notable differences in usage.

I've tested the group mode in spark-streaming-kafka-0-10, it does work.