我一直在尝试编写一个读取kafka主题的beam管道,该主题由avro记录组成。这些记录的模式可以快速更改,因此在提取相关的公共字段之前,我希望使用Confluent schema Registry来获取模式并解码事件。要么是我做错了什么,要么是文档过时了。我遵循了下面的示例:
* <p>If you want to deserialize the keys and/or values based on a schema available in Confluent
* Schema Registry, KafkaIO can fetch this schema from a
我已经创建了一个Mapper & Reducer,它使用AVRO作为输入,map-output和reduce输出。在创建MRUnit测试时,我得到了以下堆栈跟踪:
java.lang.NullPointerException
at org.apache.hadoop.io.serializer.SerializationFactory.getSerializer(SerializationFactory.java:73)
at org.apache.hadoop.mrunit.mock.MockOutputCollector.deepCopy(MockOutputCollector.j
我使用Apache Kafka与Avro序列化,使用特定的格式。我正在尝试创建自己的自定义类,并将其用作kafka消息值。但是当我尝试发送消息时,我得到了以下异常:
Exception in thread "main" org.apache.avro.AvroRuntimeException: not open
at org.apache.avro.file.DataFileWriter.assertOpen(DataFileWriter.java:82)
at org.apache.avro.file.DataFileWriter.append(DataFil
我正在尝试发送avro记录到一个卡夫卡主题使用卡夫卡生产者。我有一个User类,我正在发送该类的对象。如果我使用avroRecord.put();来设置每个属性,下面的代码可以很好地工作。但我想要的是从一个对象创建一个通用记录,而不使用avroRecord.put();为每个属性。
用户类
public class User {
int id;
String name;
public User(int id, String name) {
super();
this.id = id;
this.name = na
我已经编写了一个使用者来使用模式注册表读取Avro的通用记录。
FlinkKafkaConsumer010 kafkaConsumer010 = new FlinkKafkaConsumer010(KAFKA_TOPICS,
new KafkaGenericAvroDeserializationSchema(schemaRegistryUrl),
properties);
反序列化类如下所示:
public class KafkaGenericAvroDeserializationSchema implements KeyedDe
嗨,我们正在尝试在同一主题上有多个事件。我们使用自定义命名策略为同一主题生成不同的模式。但是,在使用者端,如果avro文件不在使用者上,则无法反序列化架构。现在,我们必须为不同的事件类型添加所有avro模式,即使使用者只关注一种事件类型(avro模式)。我们希望避免这种情况,并想知道实现这一目标的最佳方式是什么。
如果我们只添加相关模式,即没有为同一主题添加所有事件类型模式,则会得到此异常。
java.lang.IllegalStateException: This error handler cannot process 'SerializationException's
我的客户类已经使用maven-avro plugin.When创建了,我尝试运行这个程序,我得到的是Exception in thread "main" java.lang.IllegalStateException: Expecting type to be a PojoTypeInfo错误。
[main] INFO org.apache.flink.api.java.typeutils.TypeExtractor - class com.example.Customer does not contain a setter for field first_name
[main
我尝试将我的Flink应用程序部署到AWS中。此应用程序使用Apache进行反序列化/序列化传入消息。我的应用程序在本地机器上运行良好,但是当我将它部署到AWS时,我有异常(在CloudWatch日志中):Caused by: java.io.InvalidClassException: org.apache.avro.specific.SpecificRecordBase; local class incompatible: stream classdesc serialVersionUID = 4445917349737100331, local class serialVersionUI
我正在尝试创建一个Kafka Streams应用程序来处理Avro记录,但我收到以下错误:
Exception in thread "streams-application-c8031218-8de9-4d55-a5d0-81c30051a829-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather hav
我想使用avro消息并反序列化它们,而不使用Confluent模式注册表。我在本地有这个模式。因此,我只遵循了这个面向消费者的https://medium.com/@mailshine/apache-avro-quick-example-in-kafka-7b2909396c02部分。现在,我想在application.properties文件中配置这个反序列化程序( Spring boot的方式)。我尝试添加spring.kafka.consumer.value-deserializer=com.example.AvroDeserializer,但结果出现错误:“找不到com.exampl
当我试图用NullPointerException编写Avro对象时,我正在尝试发布Avro (到Kafka中)并获得一个BinaryEncoder。
下面是简化的堆栈跟踪:
java.lang.NullPointerException: null of array of com.mycode.DeeplyNestedObject of array of com.mycode.NestedObject of union of com.mycode.ParentObject
at org.apache.avro.generic.GenericDatumWriter.npe(Generic
我正在尝试从Java类层次结构创建一个avro模式。在深入研究完整结构之前,我尝试为一个简单的java类生成一个Avro模式,如下所示
package TestAvroTools;
import java.util.ArrayList;
public class InduceSchemaIPClass {
private int inta;
private String stringb;
private ArrayList<String> a = null;
private boolean booleana;
private st
我需要在KStream中使用transform()操作,但在没有设置必要的serdes时得到通常的ClassNotFoundException:
Caused by: java.lang.ClassCastException: xxx.SomeKey cannot be cast to [B
at org.apache.kafka.common.serialization.ByteArraySerializer.serialize(ByteArraySerializer.java:21)
at org.apache.kafka.common.serialization.Seri