如何使用直接流在 Kafka Spark Streaming 中指定消费者组答案

【问题标题】：how to specify consumer group in Kafka Spark Streaming using direct stream如何使用直接流在 Kafka Spark Streaming 中指定消费者组
【发布时间】：2016-07-30 05:53:48
【问题描述】：

如何使用直接流 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");

【问题讨论】：

应该和你做的完全一样。
not sure if missing something 是什么意思？请提出一个具体问题。 我尝试 X 使用库 Z 实现 Y，但使用堆栈跟踪 S 得到异常 E 之类的东西适用于 StackOverflow。
@DebosmitRay 我尝试使用 spark kafka 直接流指定消费者组的“group.id”。没有得到任何异常，但想知道这是否是在使用 createDirectStream API 方法时指定消费者组的正确方法。现在有用吗？？？

标签： java apache-spark apache-kafka spark-streaming kafka-consumer-api

【解决方案1】：

spark-streaming-kafka-0-8 中的createDirectStream 不支持组模式，因为它使用的是低级 Kafka API。

但是spark-streaming-kafka-0-10支持群组模式。

Consumer Configs

在 0.9.0.0 中，我们引入了新的 Java 消费者来替代旧的基于 Scala 的简单和高级消费者。新老消费者的配置如下所述。

在New Consumer Configs 中有group.id 项。

Spark Streaming integration for Kafka 0.10 正在使用新的 API。 https://spark.apache.org/docs/2.1.1/streaming-kafka-0-10-integration.html

Kafka 0.10 的 Spark Streaming 集成在设计上类似于 0.8 Direct Stream 方法。它提供简单的并行性，Kafka 分区和 Spark 分区之间的 1:1 对应关系，以及对偏移量和元数据的访问。但是，由于较新的集成使用新的 Kafka 消费者 API 而不是简单的 API，因此在使用上存在显着差异。

我已经在spark-streaming-kafka-0-10 中测试了组模式，它确实有效。

【讨论】：

【解决方案2】：

直接流 API 使用低级别的 Kafka API，因此无论如何都不使用消费者组。如果您想通过 Spark Streaming 使用消费者组，则必须使用基于接收器的 API。

Full details are available in the doc !

【讨论】：

我仍然可以使用直接流 API 通过多个应用程序使用数据吗？不会对每个应用程序造成任何数据丢失？