我需要从多个 Kafka 主题[基于 Avro] 进行流式传输，并将它们放入 Greenplum 中，并对有效负载进行少量修改。
Kaka 主题在配置文件中定义为一个列表，每个 Kafka 主题将有一个目标表。
我正在寻找单个 Spark 结构化应用程序和配置文件中的更新以收听新主题或停止。听主题。
我正在寻求帮助，因为我对使用单个查询与多个查询感到困惑：
val query1 = df.writeStream.start()
val query2 = df.writeStream.start()
spark.streams.awaitAnyTermination()
或
df.writeStream.start().awaitAnyTermination()
在哪些用例中应该使用多个查询而不是单个查询
【问题讨论】：
标签： scala apache-spark apache-spark-sql spark-streaming spark-structured-streaming
显然，您可以使用正则表达式模式来使用来自不同 kafka 主题的数据。
假设您有“topic-ingestion1”、“topic-ingestion2”之类的主题名称 - 然后您可以创建一个正则表达式模式来使用以“*ingestion”结尾的所有主题的数据。
一旦以您的正则表达式模式的格式创建新主题 - spark 将自动开始从新创建的主题流式传输数据。
参考： [https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#consumer-caching]
您可以使用此参数来指定您的缓存超时。 “spark.kafka.consumer.cache.timeout”。
来自火花文档：
spark.kafka.consumer.cache.timeout - 最短时间 消费者在有资格被驱逐之前可能会在池中闲置 由驱逐者。
假设您有多个接收器从 kafka 读取数据并将其写入两个不同的位置，例如 hdfs 和 hbase - 那么您可以将应用程序逻辑分支到两个 writeStreams。
如果接收器 (Greenplum) 支持批处理操作模式 - 那么您可以查看 spark 结构化流中的 forEachBatch() 函数。这将允许我们为这两个操作重用相同的 batchDF。
参考： [https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#consumer-caching]
【讨论】：