(DEPRECATED) Apache Flink User Mailing List archive.

Serialization and Deserialization of Avro messages stored in Kafka

Classic

List

Threaded

3 messages Options

Filipe Couto

Serialization and Deserialization of Avro messages stored in Kafka

Hello,

I have a few topics that I want to read from Kafka, which consist mainly on a key value pair of: timestamp (key) and value (byte array).

The bite array doesn't really have a class to deserialize from, since the Avro Record we have comes from a "SELECT * FROM..." that selects several SQL tables and in each topic we have that table represented.

We're using a GenericRecord, and since we know the structure of the table via the name of the topic we know the column names, like this: genericRecord.get("COLUMN_NAME").toString()

Given this, we're now trying to read a Kafka topic using Flink, and we have this:

The environment is the StreamExecutionEnvironment and the properties are about the Kafka serialization and deserialization and Kafka and Zookeeper IP addresses.

class...

DataStream<Object> messageStream = environment

.addSource(new FlinkKafkaConsumer010<>(baseTopic, new MyDeserializationSchema(schema), properties));

messageStream.print();

try {

environment.execute();

} catch (Exception e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

return false;

}

class MyDeserializationSchema<T> implements DeserializationSchema<T> {

private static final Logger log = LoggerFactory.getLogger(MyDeserializationSchema.class);

private final Class<T> avrotype = (Class<T>) org.apache.avro.generic.GenericRecord.class;

private final Schema schema;

public MyDeserializationSchema(Schema schema) {

this.schema = schema;

}

@Override

public T deserialize(byte[] arg0) throws IOException {

log.info("Starting deserialization");

GenericRecord genericRecord;

Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs

.toBinary(schema);

log.info(recordInjection.toString());

genericRecord = recordInjection.invert(arg0).get();

log.info(genericRecord.toString());

return (T) genericRecord;

}

@Override

public boolean isEndOfStream(T nextElement) {

return false;

}

@Override

public TypeInformation<T> getProducedType() {

return TypeExtractor.getForClass(avrotype);

}

Executing this on our server generates the following:

[2018-03-02 15:59:37,111] WARN Ignoring configured key DeSerializer (key.deserializer) (org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09)

Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: The implementation of the FlinkKafkaConsumer09 is not serializable. The object probably contains or references non serializable fields.

at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)

at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1548)

at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1460)

at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1404)

at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1386)

at com.i2s.analytics.flink.executors.LKTreatyExecutor.execute(LKTreatyExecutor.java:153)

at com.i2s.analytics.flink.job.DependenciesConsumer.main(DependenciesConsumer.java:66)

Caused by: java.io.NotSerializableException: org.apache.avro.Schema$RecordSchema

at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)

at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)

at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)

at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)

at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)