如何使用直接流在Kafka Spark流中指定使用者组 [英] how to specify consumer group in Kafka Spark Streaming using direct stream

问题描述

如何使用直接流API为kafka spark流指定消费者组ID.

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
    );

尽管我已指定配置，但不确定是否缺少某些内容.使用spark1.3

kafkaParams.put("group.id", "app1");

解决方案

直接流API使用底层的Kafka API，因此无论如何都不会使用使用者组.如果要将消费者组与Spark Streaming一起使用，则必须使用基于接收者的API.

文档中提供了完整的详细信息！

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");

解决方案

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 !

这篇关于如何使用直接流在Kafka Spark流中指定使用者组的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！