如何使用 subscribePattern 订阅新主题？答案

【问题标题】：How to subscribe to a new topic with subscribePattern?如何使用 subscribePattern 订阅新主题？
【发布时间】：2021-03-28 18:47:32
【问题描述】：

我正在使用带有 Kafka 的 Spark 结构化流，并且主题已被订阅为模式：

option("subscribePattern", "topic.*")

// Subscribe to a pattern
val df = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribePattern", "topic.*")
  .load()

一旦我开始工作并列出了一个新主题，比如topic.new_topic，该工作就不会自动开始收听新主题，它需要重新启动。

有没有办法在不重新启动作业的情况下自动订阅新模式？

火花：3.0.0

【问题讨论】：

自 3.0.0 起，消费者缓存可用，这可能允许我们从新主题中读取数据 - spark.apache.org/docs/latest/…

标签： apache-spark apache-kafka spark-structured-streaming spark-kafka-integration

【解决方案1】：

KafkaConsumer 的默认行为是每 5 分钟检查一次是否有新的分区要使用。这个配置是通过Consumer config设置的

metadata.max.age.ms：以毫秒为单位的时间段，在此之后我们强制刷新元数据，即使我们没有看到任何分区领导层更改以主动发现任何新代理或分区。

根据Kafka Specific Configuration 上的 Spark + Kafka 集成指南，您可以使用前缀 kafka. 设置此配置，如下所示：

.option("kafka.metadata.max.age.ms", "1000")

通过此设置，新创建的主题将在创建后 1 秒被消耗。

（使用 Spark 3.0.0 和 Kafka Broker 2.5.0 测试）

【讨论】：