我在Spark流中使用 kafkaUtils.createDirectStream（）。 这种方法将帮助我创建一个组的kafka使用者（作为参数传递的组ID）。该小组订阅了多个主题。每个主题都有多个分区。  这种结构看起来确实很脆弱，因为当我手动关闭一些kafka经纪人，使某些主题分区不可用时，使用者将被阻止！这导致整个消费过程受阻！  为什么 createDirectStream（）不创建多个使用方，以便当某些分区的某些使用方被阻止时，该组中的其他使用方仍可以接收消息，从而不会阻止整个流传输？

在Spark Streaming + Kafka Integration Guide 中，

  

直接流方法。它提供了简单的并行性，Kafka分区和Spark分区之间的1：1对应关系，以及>偏移和元数据的访问。

根据此描述，我想createDirectStream（）将创建多个使用者，一个分区的一个使用者，这将导致高可用性和并发性。但是，当我使用kafka cml脚本检查使用者组状态时，我注意到仅创建了一个使用者！当我关闭一些代理以使一个分区不可用时，整个流式传输过程将被阻塞！可以使用createDirectStream（）在一个组中创建多个使用者吗？

这是我使用createDirectStream的方式

val kafkaParams = Map[String,String] (
      "bootstrap.servers" -> brokers,"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer","key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer","auto.offset.reset" -> "latest","enable.auto.commit" -> "false","group.id" -> groupId
    )
    val locationStrategy = LocationStrategies.PreferConsistent
    val consumerStrategy = ConsumerStrategies.Subscribe[String,String](topicSet,kafkaParams)
    val messages = KafkaUtils.createDirectStream(ssc,locationStrategy,consumerStrategy)

这是我检查消费群体状态的方法：

 bin/kafka-consumer-groups.sh --bootstrap-server 192.168.103.236:9092,192.168.103.237:9092,[B192.168.103.238:9092 --describe --group spark_streaming_group_test

我发现只创建了一个消费者：

但这不是我想的！我假设消费者客户端从分区0和1继续接收主题ub_read_log的消息，因为它们可用！。另外，我想其他主题的消息也不会被屏蔽。 但是，一切都被封锁了！我认为这是因为createDirectStream（）仅创建一个使用者。如果创建了多个消费者，一切都会好起来的，不是吗？ 我的观察有什么问题吗？欢迎任何建议，非常感谢！