Spark Structured Streaming 与基于 Avro 的消息的 Schema Registry 集成

Question

我们有一个用例，我们尝试使用与 Schema 注册表集成的多个 Kafka 主题（AVRO 消息）。 我们正在使用 Spark 结构化流（Spark 版本：2.4.4）、Confluent Kafka（库版本：5.4.1）：

val kafkaDF: DataFrame = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<bootstrap.server>:9092")
  .option("subscribe", [List of Topics]) // multi topic subscribe
  .load()

我们正在从上面的 DF 中选择值并使用 Schema 反序列化 AVRO 消息

val tableDF = kafkaDF.select(from_avro(col("value"), jsonSchema).as("table")).select("*")

这里的障碍是，由于我们使用多个 Kafka 主题，我们已经将所有 JSON 模式集成到 MAP 中，key 是 主题名称 和 values 是各自的架构。 我们如何在上面的选择中使用相同的地图进行查找？我们尝试了 UDF，但它返回一个“col”类型，但 jsonSchema 必须是 String 类型。此外，不同主题的架构也不同

几个附加问题：

1. 上述方法对于同时使用多个主题是否正确？
2. 我们应该为每个主题使用单个消费者吗？
3. 如果我们有更多的主题，我们如何实现并行主题处理，cos 顺序可能需要大量时间。

Answer 1

如果不检查所有内容，您似乎对 from_avro 等 from_json 等的基础知识没问题。

1. https://sparkbyexamples.com/spark/spark-streaming-consume-and-produce-kafka-messages-in-avro-format/ 可以指导您完成第一部分。这也很好https://www.waitingforcode.com/apache-spark-structured-streaming/two-topics-two-schemas-one-subscription-apache-spark-structured-streaming/read#filter_topic。

2. 我会做桌子。*

3. 多个、多个模式 -->从 .avsc 读取多个此类版本或自己编写代码。

4. 对于在 Spark Streaming 应用程序中使用多个主题，问题是每个应用程序有多少个主题？没有硬性规则，除了大消费与小消费等明显的规则，如果订单很重要。可以放弃执行器资源。

5. 然后你需要像这样单独处理所有主题 - Filter on Topic - 你可以填写详细信息，因为我有点匆忙 - 使用 foreachBatch 范式。

6. 不确定您是如何将静态数据写出的，但问题不在于那个。

与此类似，则处理多个Topic：

...
... // Need to get Topic first
stream.toDS()
      .select($"topic", $"value")
      .writeStream
      .foreachBatch((dataset, _) => {
         dataset.persist() // need this

         dataset.filter($"topic" === topicA)
                .select(from_avro(col("value"), jsonSchemaTA)
                .as("tA"))
                .select("tA.*")
                .write.format(...).save(...)

         dataset.filter($"topic" === topicB)
                .select(from_avro(col("value"), jsonSchemaTB)
                .as("tB"))
                .select("tB.*")
                .write.format(...).save(...)
          ...
          ...
         dataset.unpersist()
         ...
         })

         .start().awaitTermination()

但要融入这个出色的答案：Integrating Spark Structured Streaming with the Confluent Schema Registry