java - KafkaAvroSerializer 用于在没有 schema.registry.url 的情况下序列化 Avro

Question

我是 Kafka 和 Avro 的菜鸟。所以我一直在尝试让生产者/消费者运行。到目前为止，我已经能够使用以下方法生成和使用简单的字节和字符串:生产者的配置:

    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

    Schema.Parser parser = new Schema.Parser();
    Schema schema = parser.parse(USER_SCHEMA);
    Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);

    KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);

    for (int i = 0; i < 1000; i++) {
        GenericData.Record avroRecord = new GenericData.Record(schema);
        avroRecord.put("str1", "Str 1-" + i);
        avroRecord.put("str2", "Str 2-" + i);
        avroRecord.put("int1", i);

        byte[] bytes = recordInjection.apply(avroRecord);

        ProducerRecord<String, byte[]> record = new ProducerRecord<>("mytopic", bytes);
        producer.send(record);
        Thread.sleep(250);
    }
    producer.close();
}

现在一切都很好，但当我尝试序列化 POJO 时，问题就来了。因此，我能够使用 Avro 提供的实用程序从 POJO 获取 AvroSchema。对模式进行硬编码，然后尝试创建一个通用记录以通过 KafkaProducer 发送生产者现在设置为:

    Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.KafkaAvroSerializer");

Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(USER_SCHEMA); // this is the Generated AvroSchema
KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);

这就是问题所在:当我使用 KafkaAvroSerializer 时，生产者没有出现是因为:缺少强制参数:schema.registry.url

我仔细阅读了为什么需要这样做，以便我的消费者能够破译生产者发送给我的任何内容。但是模式不是已经嵌入到 AvroMessage 中了吗？如果有人可以分享一个使用 KafkaProducer 和 KafkaAvroSerializer 而无需指定 schema.registry.url 的工作示例，那就太好了

也非常感谢有关模式注册表实用程序的任何见解/资源。

谢谢!

Answer 1

先注意:KafkaAvroSerializer vanilla apache kafka 中未提供 - 它由 Confluent Platform 提供。 ( https://www.confluent.io/ )，作为其开源组件 ( http://docs.confluent.io/current/platform.html#confluent-schema-registry ) 的一部分

快速回答:不，如果您使用 KafkaAvroSerializer ，您将需要一个架构注册表。在此处查看一些示例: http://docs.confluent.io/current/schema-registry/docs/serializer-formatter.html

模式注册表的基本思想是每个主题将引用一个avro模式(即，您只能发送彼此一致的数据。但是一个模式可以有多个版本，所以您仍然需要识别每条记录的架构)

我们不想像您暗示的那样为每个数据编写架构 - 通常，架构比您的数据大!每次阅读时都解析它会浪费时间，并且会浪费资源(网络，磁盘，cpu)

相反，模式注册表实例将进行绑定(bind) avro schema <-> int schemaId然后序列化程序将只在数据之前写入这个 id，从注册表中获取它(并将其缓存以备后用)。

所以在kafka里面，你的记录会是[<id> <bytesavro>] (和技术原因的魔术字节)，这是只有 5 个字节的开销(与您的模式的大小进行比较)在阅读时，您的用户会找到与 id 对应的模式，以及与之相关的反序列化器 avro 字节。您可以在 confluent doc 中找到更多方法

如果你真的想为每条记录编写模式，你将需要一个其他的序列化程序(我想写你自己的，但这很容易，只需重用 https://github.com/confluentinc/schema-registry/blob/master/avro-serializer/src/main/java/io/confluent/kafka/serializers/AbstractKafkaAvroSerializer.java 并删除模式注册表部分用模式替换它，阅读相同)。但是如果你使用 avro，我真的不鼓励这样做 - 有一天，你将需要实现类似 avro registry 的东西来管理版本控制