[FLINK-5039] Avro GenericRecord support is broken - ASF JIRA

Attach files

Attach Screenshot

Voters

Watch issue

Watchers

Create sub-task

Link

Clone

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Bug
Status: Closed
Priority: Blocker
Resolution: Fixed
Affects Version/s: 1.1.3
Fix Version/s: 1.1.4, 1.2.0
Component/s: API / DataSet
Labels:
None

Description

Avro GenericRecord support was introduced in ~~FLINK-3691~~, but it seems like the GenericRecords are not properly (de)serialized.

This can be easily seen with a program like this:

  env.createInput(new AvroInputFormat<>(new Path("somefile.avro"), GenericRecord.class))
    .first(10)
    .print();

which will print records in which all fields have the same value:

{"foo": 1478628723066, "bar": 1478628723066, "baz": 1478628723066, ...}
{"foo": 1478628723179, "bar": 1478628723179, "baz": 1478628723179, ...}

If I'm not mistaken, the AvroInputFormat does essentially TypeExtractor.getForClass(GenericRecord.class), but GenericRecords are not POJOs.

Furthermore, each GenericRecord contains a pointer to the record schema. I guess the current naive approach will serialize this schema with each record, which is quite inefficient (the schema is typically more complex and much larger than the data). We probably need a TypeInformation and TypeSerializer specific to Avro GenericRecords, which could just use avro serialization.