为什么将 Avro 与 Kafka 结合使用 - 如何处理 POJO

我有一个 Spring 应用程序，它是我的 kafka 生产者，我想知道为什么 avro 是最好的选择。 我读到了它以及它提供的所有内容，但为什么我不能序列化我用 jackson 自己创建的 POJO 并将其发送到 kafka？

我这样说是因为 avro 生成的 POJO 并不那么直接。 最重要的是，它需要 maven 插件和 .avsc 文件。

例如，我在我的 kafka 生产者上创建了一个名为 User 的 POJO：

public class User {

    private long    userId;

    private String  name;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public long getUserId() {
        return userId;
    }

    public void setUserId(long userId) {
        this.userId = userId;
    }

}

我将其序列化并将其发送到我在 kafka 中的用户主题。然后我有一个消费者，它本身有一个 POJO 用户并反序列化消息。 是空间的问题吗？这样序列化和反序列化不是也更快吗？更不用说维护模式注册表的开销了。

你不需要AVSC，您可以使用 AVDL 文件 https://avro.apache.org/docs/1.8.1/idl.html，它基本上看起来与只有字段的 POJO 相同

@namespace("com.example.mycode.avro")
protocol ExampleProtocol {
   record User {
     long id;
     string name;
   }
}

其中，当使用idl-protocolMaven 插件的目标是为您创建这个 AVSC，而不是您自己编写。

{
  "type" : "record",
  "name" : "User",
  "namespace" : "com.example.mycode.avro",
  "fields" : [ {
    "name" : "id",
    "type" : "long"
  }, {
    "name" : "name",
    "type" : "string"
  } ]
}

它还会放置一个SpecificData POJO User.java在您的类路径中以便在代码中使用。

如果您已有 POJO，则无需使用 AVSC 或 AVDL 文件。有一些库可以转换 POJO。例如，你can use杰克逊 https://github.com/FasterXML/jackson-dataformats-binary/tree/master/avro#generating-avro-schema-from-pojo-definition，这不仅适用于 JSON，您可能只需要创建一个JacksonAvroSerializer例如，对于 Kafka，或者查找是否存在。

阿芙罗还有基于反射的内置库 https://blog.jeffli.me/blog/2014/04/05/avro-cookbook-part-ii/#parse-schema-from-existing-java-class/.

Confluence SchemaRegistry 序列化器有一个使用基于反射的模型的设置。

所以对于这个问题——为什么选择 Avro（卡夫卡）？

嗯，拥有一个模式就是好东西。想想 RDBMS 表，您可以解释该表，并且可以看到所有列。转向 NoSQL 文档数据库，它们几乎可以包含任何内容，这就是 Kafka 的 JSON 世界。

假设您的 Kafka 集群中有消费者，他们不知道主题中的内容，他们必须确切地知道主题中产生了谁/什么。他们可以尝试控制台消费者，如果它是像 JSON 这样的明文，那么他们必须找出一些他们感兴趣的字段，然后执行类似 HashMap 的片状操作.get("name")一次又一次的操作，只是在字段不存在时遇到NPE。有了 Avro，您明确定义默认值和可为空的字段。

你不是required使用模式注册表，但它提供了这种类型explain topicRDBMS 类比的语义。它还使您无需将架构与每条消息一起发送，以及 Kafka 主题上的额外带宽费用。不过，该注册表不仅对 Kafka 有用，它还可以用于 Spark、Flink、Hive 等，用于围绕流数据摄取的所有数据科学分析。

假设您确实想使用 JSON，那么尝试使用 MsgPack 代替 https://msgpack.org/您可能会看到 Kafka 吞吐量有所增加，并节省了代理上的磁盘空间

您还可以使用其他格式，例如 Protobuf 或 Thrift，正如 Uber 所比较的那样 https://eng.uber.com/trip-data-squeeze/