如何从apache nifi在kafka主题中产生Avro消息，然后使用kafka流阅读它？

Question

我想使用apache nifi将一些通用数据生成到kafka主题中，并且我希望这些数据是avro格式的。我为它所做的：

1. 在架构注册表中创建新架构:

{“type”：“record”,“name”：“my_schema”,“namespace”：“my_namespace”,“doc”：“”,“fields”：[{“name”：“key”,“type”：“int”},{“name”：“value”,“type”：[“null”,“int”]},{“name”：“event_time”,“type”：“long”}]}

然后我试着用Kafka的Streams来读它：

公共类测试{private final static Logger Logger=Logger.getlogger（kafkafilterusingcacheavro.class）；

public static void main(String[] args) {
    Properties properties = new Properties();

    properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
    properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "app");
    properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
    properties.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "registry:8081");

    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, GenericRecord> source = builder.stream("topic");
    source.foreach((k, v) -> logger.info(String.format("[%s]: %s", k, v.toString())));

    Topology topology = builder.build();
    KafkaStreams streams = new KafkaStreams(topology, properties);
    streams.start();
}

}

GenericAvroSerde-https://html" target="_blank">github.com/johnreedlol/kafka-streams/blob/master/src/main/Java/io/confluent/examples/streams/utils/GenericAvroSerde.Java

结果我得到了错误：

原因：org.apache.kafka.Common.Errors.SerializationException：反序列化id-1的Avro消息时出错，原因：org.apache.kafka.Common.Errors.SerializationException：未知的魔术字节！

我还尝试在AvroReader\Writer中显式设置avro模式，但没有帮助。另外，如果我尝试简单地从topic读取字节并将其转换为字符串表示，我会得到如下结果：

objavro.schema{“type”：“record”,“name”：“my_schema”,“namespace”：“my_namespace”,“doc”：“”,“fields”：[{“name”：“key”,“type”：“int”},{“name”：“value”,“type”：[“null”,“int”]},{“name”：“event_time”,“type”：“long”}]}avro.codecsnappy[4]m[©Q:àg0]“/}1/2{[4]m[©Q:àg0

我该怎么修好它？

Answer 1

在PublishKafka处理器中，Avro writer配置有“Embedded Avro Schema”的“Schema Write Strategy”。这意味着写入Kafka的消息是嵌入了完整模式的标准Avro消息。

在您的消费者端（Kafka流）上，它似乎希望使用汇合模式注册表，在这种情况下，它希望的不是嵌入式Avro模式，而是指定模式id的特殊字节序列，然后是裸Avro消息。

假设您希望保持您的使用者原样，那么在NiFi方面，您将希望将Avro Writer的“模式编写策略”更改为“合流模式注册表引用”。我认为这可能还需要您更改Avro reader以使用合流模式注册服务访问模式。

或者，也许有一种方法可以使Kafka流读取嵌入式模式，而不使用汇流模式注册表，但是我以前没有使用过Kafka流，所以我不能说这是否可行。