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

Question

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

此外，通过查看 apache/spark repo https://github.com/apache/spark/blob/master/docs/structured-streaming-kafka-integration.md 的 master 分支的文档，我们可以了解到此类功能旨在在以后的 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

在 master 分支中你可以找到修改，可以设置 prefix 或特定的 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 - 设置之前生成的 groupId，如果 kafka.group.id 没有在属性中传递

Answer 2

自 Spark 3.0.0

根据Structured Kafka Integration Guide，您可以提供ConsumerGroup 作为选项kafka.group.id：

val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .option("kafka.group.id", "myConsumerGroup")
  .load()

但是，Spark 不会提交任何偏移量，因此您的 ConsumerGroups 的偏移量不会存储在 Kafka 的内部主题 __consumer_offsets 中，而是存储在 Spark 的检查点文件中。

能够设置group.id 意味着处理Kafka 的最新功能Authorization using Role-Based Access Control，您的ConsumerGroup 通常需要遵循命名约定。

讨论并解决了一个 Spark 3.x 应用程序设置kafka.group.id 的完整示例here。

Answer 3

现在有了 spark3.0，你可以为 kafka 指定 group.id https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#kafka-specific-configurations

Answer 4

Structured Streaming guide 似乎对此非常明确：

请注意，以下 Kafka 参数 无法设置，而 Kafka source 或 sink 会抛出异常：

group.id：Kafka 源将为每个查询创建一个唯一的组 id 自动地。

auto.offset.reset：设置源选项 startingOffsets 指定从哪里开始。