为什么两个火花流作业从具有相同组id的相同Kafka主题拉消息不平衡负载，但得到相同的消息？

Question

Kafka 0.8官方文档对Kafka消费者描述如下：

“消费者用一个消费者组名称给自己贴标签，发布到主题的每条消息都被传递到每个订阅消费者组中的一个消费者实例。消费者实例可以在不同的进程中或在不同的机器上。如果所有消费者实例都有相同的消费者组，那么这就像传统的队列平衡消费者的负载一样。”

我用Kafka0.8.1.1设置一个Kafka集群，并使用Spark Streaming作业（Spark 1.3）从其主题中提取数据。Spark Streaming代码如下：

    ... ...

    HashMap<String, String> kafkaParams = new HashMap<String, String>();
    kafkaParams.put("metadata.broker.list", brokerList);
    kafkaParams.put("group.id", groupId);

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

    messages.foreachRDD(new Function<JavaPairRDD<String, String>, Void>() {

        @Override
        public Void call(JavaPairRDD<String, String> rdd) throws Exception {
            long msgNum = strJavaRDD.count();
            System.out.println("There are " + msgNum + " messages read from Kafka.");

        ... ...

        return null;}});

然后我提交了两个Spark流媒体作业，以访问具有相同组id的同一主题。我假设当我向该主题发送100条消息时，这两个作业总共收到100条消息（例如job1 get 50和job2 get 50；或者job1 get 100和job2 get 0）。然而，他们分别得到了100分。这样的结果似乎与Kafka博士所说的有所不同。

我的密码有什么问题吗？我是否正确设置了组id配置？这是createDirectStream（）的错误还是设计？

测试环境：Kafka0.8.1.1Spark 1.3.1

Answer 1

创建两个不同的spark应用程序来用相同的消息做相同的事情是没有意义的。使用一个有更多执行者的应用程序。

Answer 2

组是Kafka0.9版之前的高级消费API的一个功能，它在简单消费API中不可用createDirectStream使用简单的消费者API。

一些提示：

>

使用简单消费者实现的主要原因是，您希望对分区消耗的控制比消费者组提供的更大。（例如：多次阅读一条消息）

CreateDirectStream：这种方法不是使用接收器来接收数据，而是定期查询Kafka每个主题分区中的最新偏移量，并相应地定义每个批处理中要处理的偏移量范围。

参考：

1. Spark Streaming Kafka集成指南

Kafka 0.9.0版本添加了一个新的Java使用者，以取代现有的基于ZooKeeper的高级使用者和低级使用者API。然后您可以使用group，同时提交偏移量手册。