apache-spark - 如何在结构化流中为kafka数据源中的消费者组设置group.id？

Question

我想使用 Spark 结构化流从安全的 kafka 中读取数据。这意味着我需要强制使用特定的 group.id。但是，正如文档中所述，这是不可能的。尽管如此，在databricks文档中https://docs.azuredatabricks.net/spark/latest/structured-streaming/kafka.html#using-ssl ，它说这是可能的。这仅指azure集群吗？

此外，通过查看 apache/spark 存储库的 master 分支的文档 https://github.com/apache/spark/blob/master/docs/structured-streaming-kafka-integration.md ，我们可以理解这些功能是打算在以后的 Spark 版本中添加的。您知道有这样一个稳定版本的计划吗？该版本将允许设置消费者 group.id？

如果没有，Spark 2.4.0 是否有任何解决方法可以设置特定的消费者 group.id？

Answer 1

目前(v2.4.0)这是不可能的。

您可以在 Apache Spark 项目中检查以下行:

https://github.com/apache/spark/blob/v2.4.0/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala#L81 - 生成group.id

https://github.com/apache/spark/blob/v2.4.0/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala#L534 - 在用于创建KafkaConsumer的属性中设置它

在主分支中，您可以找到修改，可以设置前缀或特定group.id

https://github.com/apache/spark/blob/master/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala#L83 - 根据组前缀(groupidprefix)生成group.id

https://github.com/apache/spark/blob/master/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala#L543 - 如果未在属性中传递 kafka.group.id，则设置先前生成的 groupId