当一个主题有多个主题时，如何处理来自 Kafka使用 Apache Beam的 Avro 输入？

如何解决当一个主题有多个主题时，如何处理来自 Kafka使用 Apache Beam的 Avro 输入？

为了使用 KafkaIO 通过 Apache Beam 处理 Avro 编码的消息，需要传递一个 ConfluentSchemaRegistryDeserializerProvider 实例作为值解串器。

一个典型的例子是这样的：

PCollection<KafkaRecord<Long,GenericRecord>> input = pipeline
  .apply(KafkaIO.<Long,GenericRecord>read()
     .withBootstrapServers("kafka-broker:9092")
     .withTopic("my_topic")
     .withKeyDeserializer(LongDeserializer.class)
     .withValueDeserializer(
         ConfluentSchemaRegistryDeserializerProvider.of("http://my-local-schema-registry:8081","my_subject"))

但是，我想使用的一些 Kafka 主题有多个不同的主题（事件类型）（出于排序原因）。因此，我无法提前提供一个固定的主题名称。如何解决这个困境？

（我的目标是最终使用 BigQueryIO 将这些事件推送到云端。）

解决方法

您可以多次阅读，每个主题一次，然后Flatten。