问题描述
为了使用 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。