2个具有相同消费者组id的spark stream作业

Question

我正在尝试对消费者群体进行实验

这是我的代码片段

public final class App {

private static final int INTERVAL = 5000;

public static void main(String[] args) throws Exception {

    Map<String, Object> kafkaParams = new HashMap<>();
    kafkaParams.put("bootstrap.servers", "xxx:9092");
    kafkaParams.put("key.deserializer", StringDeserializer.class);
    kafkaParams.put("value.deserializer", StringDeserializer.class);
    kafkaParams.put("auto.offset.reset", "earliest");
    kafkaParams.put("enable.auto.commit", true);
    kafkaParams.put("auto.commit.interval.ms","1000");
    kafkaParams.put("security.protocol","SASL_PLAINTEXT");
    kafkaParams.put("sasl.kerberos.service.name","kafka");
    kafkaParams.put("retries","3");
    kafkaParams.put(GROUP_ID_CONFIG,"mygroup");
    kafkaParams.put("request.timeout.ms","210000");
    kafkaParams.put("session.timeout.ms","180000");
    kafkaParams.put("heartbeat.interval.ms","3000");
    Collection<String> topics = Arrays.asList("venkat4");

    SparkConf conf = new SparkConf();
    JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(INTERVAL));


    final JavaInputDStream<ConsumerRecord<String, String>> stream =
            KafkaUtils.createDirectStream(
                    ssc,
                    LocationStrategies.PreferConsistent(),
                    ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
            );

    stream.mapToPair(
            new PairFunction<ConsumerRecord<String, String>, String, String>() {
                @Override
                public Tuple2<String, String> call(ConsumerRecord<String, String> record) {
                    return new Tuple2<>(record.key(), record.value());
                }
            }).print();


    ssc.start();
    ssc.awaitTermination();


}

}

当我同时运行两个spark流媒体作业时，它会出错

线程“main”java中出现异常。lang.IllegalStateException:当前没有分配给组织上的分区venkat4-1。阿帕奇。Kafka。客户。消费者内部。订阅状态。组织上的assignedState（SubscriptionState.java:251）。阿帕奇。Kafka。客户。消费者内部。订阅状态。needOffsetReset（SubscriptionState.java:315）位于org。阿帕奇。Kafka。客户。消费者Kafka苏美尔。参见org上的kToEnd（KafkaConsumer.java:1170）。阿帕奇。火花流动。Kafka010。导演Kafka因普特流。LatestOffset（DirectKafkaInputDStream.scala:197）位于org。阿帕奇。火花流动。Kafka010。导演Kafka因普特流。计算（DirectKafkaInputDStream.scala:214）位于org。阿帕奇。火花流动。数据流。数据流$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7。在org上申请（DStream.scala:341）。阿帕奇。火花流动。数据流。数据流$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7。在scala上应用（DStream.scala:341）。util。动态变量。org上的withValue（DynamicVariable.scala:58）。阿帕奇。火花流动。数据流。数据流$$anonfun$getOrCompute$1$$anonfun$1。在org上申请（DStream.scala:340）。阿帕奇。火花流动。数据流。数据流$$anonfun$getOrCompute$1$$anonfun$1。在org上申请（DStream.scala:340）。阿帕奇。火花流动。数据流。数据流。在org上创建rddwithLocalProperties（DStream.scala:415）。阿帕奇。火花流动。数据流。数据流$$anonfun$getOrCompute$1。在org上申请（DStream.scala:335）。阿帕奇。火花流动。数据流。数据流$$anonfun$getOrCompute$1。在scala上应用（DStream.scala:333）。选项orElse（Option.scala:289）

https://www.wisdomjobs.com/e-university/apache-kafka-tutorial-1342/apache-kafka-consumer-group-example-19004.html创建具有相同组的kafka消费者的单独实例将创建分区的重新平衡。我相信消费者不会容忍这种重新平衡。我应该如何解决这个问题

下面是使用的命令

SPARK_KAFKA_VERSION=0.10 spark2 submit--num executors 2--master thread--deploy mode client--files jaas。conf#jaas。形态，蜂巢。键盘#蜂巢。keytab--驱动程序java选项“-Djava.security.auth.login.config=./jaas.conf”--类流。App--conf“spark.executor.extraJavaOptions=-Djava.security.auth.login.config=./jaas.conf”--conf spark。流动。Kafka。消费者隐藏物enabled=false 1-1.0-SNAPSHOT。罐子

Answer 1

@Ravikumar为延迟道歉。

我的测试是这样做的

a、 我的主题有3个分区b。spark streaming作业由2个执行器启动，运行良好。c、 后来，我决定用另一个实例来扩展它，运行另一个spark streaming作业，其中有一个执行器，以匹配失败的第三个分区。

关于你的说法：当你开始第二个火花流作业时，另一个消费者试图从同一个消费者Groupid消费同一个分区。所以它抛出了错误。是的，这是完全正确的。但是为什么它不能容忍是这里的问题。

引用您突出显示的文档：

Kafka将一个主题的分区分配给一个组中的消费者，因此每个分区只由该组中的一个消费者使用。Kafka保证，一条消息只能由组中的单个消费者阅读。每当任何代理出现故障或向现有主题添加新分区时，Kafka都会重新平衡分区存储。这是kafka特有的如何在代理中跨分区平衡数据的方法。如果添加更多进程/线程，Kafka将重新平衡。如果任何消费者或代理未能向ZooKeeper发送心跳信号，则Kafka群集可以重新配置ZooKeeper。

这也是我对火花流工作的期望。我尝试了能够忍受再平衡的普通Kafka客户。

你在文档中的观点“缓存是由topic分区和group.id键控的，所以使用

除了公关部https://github.com/apache/spark/pull/21038--下面是

“当新消费者加入消费者组以重新平衡分区时，Kafka分区可以被撤销。但当前的Spark Kafka connector代码确保没有分区撤销场景，因此尝试从撤销的分区获取最新偏移量将引发JIRA提到的异常。”

很高兴结束这个话题非常感谢你的回应

Answer 2

https://www.wisdomjobs.com/e-university/apache-kafka-tutorial-1342/apache-kafka-consumer-group-example-19004.html创建具有相同组的kafka消费者的单独实例将创建分区的重新平衡。我相信消费者不会容忍这种重新平衡。我应该如何解决这个问题

现在，所有分区都只由一个使用者使用。如果数据摄取率很高，消费者可能会以摄取的速度缓慢地消耗数据。

向同一个consumergroup添加更多消费者，以使用某个主题中的数据并提高消费率。Spark streaming使用这种方法在Kafka分区和Spark分区之间实现1:1并行。Spark将在内部处理它。

如果你有比主题分区更多的消费者，那么它将处于空闲状态，资源利用率也会降低。始终建议使用者应小于或等于分区计数。

如果添加了更多的进程/线程，Kafka将重新平衡。如果任何消费者或代理未能向ZooKeeper发送心跳，Kafka集群可以重新配置ZooKeeper。

当任何代理失败或向现有主题添加新分区时，kafka都会重新平衡分区存储。这是kafka特定的如何平衡代理中跨分区的数据。

Spark streaming在Kafka分区和Spark分区之间提供了简单的1:1并行性。如果您没有使用ConsumerStragies提供任何分区详细信息。分配、使用给定主题的所有分区。

Kafka将一个主题的分区分配给组中的消费者，这样每个分区都被组中的一个消费者使用。Kafka保证消息只被组中的单个消费者读取。

当您启动第二个spark流媒体作业时，另一个使用者尝试使用来自同一使用者groupid的同一分区。所以它抛出了错误。

val alertTopics = Array("testtopic")

val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> sparkJobConfig.kafkaBrokers,
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> sparkJobConfig.kafkaConsumerGroup,
  "auto.offset.reset" -> "latest"
)

val streamContext = new StreamingContext(sparkContext, Seconds(sparkJobConfig.streamBatchInterval.toLong))

val streamData = KafkaUtils.createDirectStream(streamContext, PreferConsistent, Subscribe[String, String](alertTopics, kafkaParams))

如果您想使用特定于分区的spark作业，请使用以下代码。

val topicPartitionsList =  List(new TopicPartition("topic",1))

val alertReqStream1 = KafkaUtils.createDirectStream(streamContext, PreferConsistent, ConsumerStrategies.Assign(topicPartitionsList, kafkaParams))

https://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html#consumerstrategies

消费者可以通过使用samegroup.id.加入一个组

val topicPartitionsList =  List(new TopicPartition("topic",3), new TopicPartition("topic",4))

    val alertReqStream2 = KafkaUtils.createDirectStream(streamContext, PreferConsistent, ConsumerStrategies.Assign(topicPartitionsList, kafkaParams))

再添加两个消费者就是添加到同一个groupid中。

请阅读Spark-Kafka集成指南。https://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html

希望这有帮助。