ApacheFlink—是否可以反序列化avro消息(使用来自kafka的消息),而不在confluentregistryavrodeserializationschema中提供读取器模式

dauxcl2d  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(428)

我在apache flink中使用kafka连接器来访问由合流kafka服务的流。
除了架构注册表url ConfluentRegistryAvroDeserializationSchema.forGeneric(...) 应为“reader”架构。我不想提供read模式,而是想使用同一个writer的模式(在注册表中查找)来读取消息,因为使用者不会有最新的模式。

  1. FlinkKafkaConsumer010<GenericRecord> myConsumer =
  2. new FlinkKafkaConsumer010<>("topic-name", ConfluentRegistryAvroDeserializationSchema.forGeneric(<reader schema goes here>, "http://host:port"), properties);
  3. myConsumer.setStartFromLatest();

https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html “使用这些反序列化架构记录将与从架构注册表检索并转换为静态提供的架构一起读取”
既然我不想在用户端保留模式定义,那么如何使用writer的模式反序列化来自kafka的avro消息呢?
谢谢你的帮助!

slmsl1lt

slmsl1lt1#

我认为不可能直接使用 ConfluentRegistryAvroDeserializationSchema.forGeneric . 它打算与读卡器模式一起使用,并且它们具有检查这一点的前提条件。
你必须实现你自己的。两件重要的事情:
specific.avro.reader 为false(否则会得到特定的记录)
这个 KafkaAvroDeserializer 必须延迟初始化(因为它本身不可序列化,因为它持有对schema registry客户机的引用)

  1. import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
  2. import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
  3. import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
  4. import io.confluent.kafka.serializers.KafkaAvroDeserializer;
  5. import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
  6. import java.util.HashMap;
  7. import java.util.Map;
  8. import org.apache.avro.generic.GenericRecord;
  9. import org.apache.flink.api.common.typeinfo.TypeInformation;
  10. import org.apache.flink.api.java.typeutils.TypeExtractor;
  11. import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
  12. public class KafkaGenericAvroDeserializationSchema
  13. implements KeyedDeserializationSchema<GenericRecord> {
  14. private final String registryUrl;
  15. private transient KafkaAvroDeserializer inner;
  16. public KafkaGenericAvroDeserializationSchema(String registryUrl) {
  17. this.registryUrl = registryUrl;
  18. }
  19. @Override
  20. public GenericRecord deserialize(
  21. byte[] messageKey, byte[] message, String topic, int partition, long offset) {
  22. checkInitialized();
  23. return (GenericRecord) inner.deserialize(topic, message);
  24. }
  25. @Override
  26. public boolean isEndOfStream(GenericRecord nextElement) {
  27. return false;
  28. }
  29. @Override
  30. public TypeInformation<GenericRecord> getProducedType() {
  31. return TypeExtractor.getForClass(GenericRecord.class);
  32. }
  33. private void checkInitialized() {
  34. if (inner == null) {
  35. Map<String, Object> props = new HashMap<>();
  36. props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, registryUrl);
  37. props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, false);
  38. SchemaRegistryClient client =
  39. new CachedSchemaRegistryClient(
  40. registryUrl, AbstractKafkaAvroSerDeConfig.MAX_SCHEMAS_PER_SUBJECT_DEFAULT);
  41. inner = new KafkaAvroDeserializer(client, props);
  42. }
  43. }
  44. }
  1. env.addSource(
  2. new FlinkKafkaConsumer<>(
  3. topic,
  4. new KafkaGenericAvroDeserializationSchema(schemaReigstryUrl),
  5. kafkaProperties));
展开查看全部

相关问题