如何在结构化流媒体中为kafka数据源中的consumer group设置group.id？

rryofs0p 于 2021-07-09 发布在 Spark

关注(0)|答案(4)|浏览(402)

我想使用spark结构化流媒体来读取一个安全的Kafka。这意味着我需要强制一个特定的group.id。然而，如文件中所述，这是不可能的。不过，在databricks文档中https://docs.azuredatabricks.net/spark/latest/structured-streaming/kafka.html#using-ssl，它说这是可能的。这是否仅指azure群集？
另外，通过查看apache/spark repo的主分支的文档https://github.com/apache/spark/blob/master/docs/structured-streaming-kafka-integration.md，我们可以理解，这样的功能将在以后的spark版本中添加。您知道这样一个稳定的版本有什么计划吗？它允许设置consumer group.id？
如果没有，spark 2.4.0是否有解决方法来设置特定的consumer group.id？

apache-kafka apache-spark spark-structured-streaming spark-kafka-integration

来源：https://stackoverflow.com/questions/66914813/multipe-consumers-with-same-consumer-group-id-in-spark-structured-streaming

4条答案

按热度按时间

fjaof16o1#

自spark 3.0.0以来

根据结构化kafka集成指南，您可以提供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不会提交回任何偏移量，因此您的ConsumerGroup的偏移量不会存储在kafka的内部主题\uu consumer\u偏移量中，而是存储在spark的检查点文件中。
能够设置 group.id 旨在使用基于角色的访问控制来处理kafka的最新功能授权，而您的consumergroup通常需要遵循命名约定。
spark 3.x应用程序设置的完整示例 kafka.group.id 在这里讨论和解决。

赞(0）回复(0）举报 2021-07-09

0g0grzrc2#

现在使用spark3.0，可以为kafka指定group.idhttps://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#kafka-特定配置

赞(0）回复(0）举报 2021-07-09

7jmck4yq3#

目前（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 -生成组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 -基于组前缀生成group.id( groupidprefix )
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 没有传入属性