Processing data without deserialization with flatbuffers

While reading about Arrow I was reminded about flatbuffers. I decided to try them for a simple case to see how they compare to hand-coded data structures. The promise was that given a schema described with an IDL similar to PBs/Thrift, the flatc compiler can generate data structures that do not require deserialization to perform operations. Which comes in handy when you have a massive input data stream.

I created a trivial schema that purports to represent a simplistic time series. It's basically a sequence of (time, value) pairs. I was curious how useful the generated code could be. For example, would it be smart enough to unwrap an array of fixed-size value pairs into two parallel arrays? 

The first nasty surprise was to learn that flatbuffers don't have an official maven artifact. They don't even have a maven plugin. This alone is a huge red flag even though some kind people came to the rescue. At least "brew install flatbuffers" actually installed the flatc compiler on MacOS. 

Flatbuffers have "struct" and "table" abstractions. The former is supposed to be bare-bones serializable data structure. The latter helps with schema changes but adds overhead to store some schema details. I was interested in the most compact format possible. My first attempt was to introduce a DataPoint struct and then have an array of them. No can do, flatc does not allow it. My second attempt was more successful. With two primitive type arrays I was actually able to compile my IDL.

To begin with, flatbuffers internally use ByteBuff instances and not byte arrays. That complicates life if you want something simple. I imagine it should complicate life even if your use case is more complex because there is no support for configurable buffer allocation. You can use a ByteBuff instance of your own but if you fail to size it appropriately, the framework will allocate a new large instance.

When I looked at the way an array is written to the ByteBuff I was disappointed to see that they copy it one element at a time. Simple omissions like that or absence of signatures such as "write(byte[] array, int from, int length)" are unexpected in a framework aimed at people who care about efficiency. It was also disappointing to see 50+ bytes of overhead when serializing a 10*16 bytes record. Those tables are not free.

So flatbuffers could still be an option if you deal with complicated, deeply-nested data structures. For simple record types they do not seem to provide any benefit but still require exorbitant effort in configuration management.

Apache Avro serialization without code generation and IDL

It is not a secret that real programmers just love writing their own serializaiton frameworks. Everyone from JBoss to GOOG has one. So it's only natural for the Hadoop stack guys to have one of their own :) What makes Avro different from Protocol Buffers and Thrift though is that one does not need IDL and code generation to send data streams between nodes of a distributed system. An Avro data stream always carries its JSON-formatted schema with it and so can be deserialized by a client without any previous knowledge of what could be inside. 

While experimenting with this feature I did not find good examples not requiring IDL and code generation so I came up with my own. What I have in this tiny test project is a couple of builders for Avro Schema and GenericRecord classes and corresponding unit tests. The tests demonstrate writing to an output stream and reading back. 

In the most basic case, an entire stream/file represents a serialized sequence of records of the same schema. All one needs to write data to an output stream is a schema and data records created with the schema. The very same generic record type is used to read data from an input stream. In the test below we just assume a certain schema but in production code one would need to check schema type of each field. Also, note the idiomatic Avro approach to reusing the same record instance to read the whole data set.

@Test
    public void testUniformSchemaSerialization() throws Exception {
        final Schema schema = new AvroSchemaBuilder(2).string(COLUMN1).int32(COLUMN2).build("DATA");

        final ByteArrayOutputStream out = new ByteArrayOutputStream();
        final DataFileWriter<GenericRecord> writer = new DataFileWriter<GenericRecord>(new GenericDatumWriter<GenericRecord>(schema)).create(schema, out);

        final AvroRecordBuilder builder = new AvroRecordBuilder(schema);
        writer.append(builder.field(CITY1).field(POPULATION1).build());
        writer.append(builder.field(CITY2).field(POPULATION2).build());
        writer.close();

        final DataFileStream<GenericRecord> reader = new DataFileStream<GenericRecord>(new ByteArrayInputStream(out.toByteArray()), new GenericDatumReader<GenericRecord>());
        GenericRecord deserialized = null;

        assertTrue(reader.hasNext());
        deserialized = reader.next(deserialized);
        assertEquals(new Utf8(CITY1), deserialized.get(COLUMN1));
        assertEquals(POPULATION1, deserialized.get(COLUMN2));

        assertTrue(reader.hasNext());
        deserialized = reader.next(deserialized);
        assertEquals(new Utf8(CITY2), deserialized.get(COLUMN1));
        assertEquals(POPULATION2, deserialized.get(COLUMN2));

        assertFalse(reader.hasNext());
    }

The generated schema looks like this:

{
  "type" : "record",
  "name" : "DATA",
  "fields" : [ {
    "name" : "CITY",
    "type" : "string"
  }, {
    "name" : "POPULATION",
    "type" : "int"
  } ]
}

In real life the need to support headers (e.g. with version-like fields) and/or footers (e.g. with checksum-like fields) of some kind is also likely to arise. In contrast to actual data tuples, header and footer formats are likely to be the same in all streams.There is one obvious way to implement it with the Avro union type. We declare the output schema for our records to be a union of three possible schemas. The ones for header and footer will have only one record each.

final String HEADER_FIELD1 = "VERSION";
        final String FOOTER_FIELD1 = "CHKSUM";
        final String HEADER_SCHEMA = "HEADER";
        final String BODY_SCHEMA = "DATA";
        final String FOOTER_SCHEMA = "FOOTER";
        final int version = 2010;
        final long total = 2;

        final Schema schema = Schema.createUnion(
                Lists.<Schema>newArrayList(
                        new AvroSchemaBuilder(1).int32(HEADER_FIELD1).build(HEADER_SCHEMA),
                        new AvroSchemaBuilder(2).string(COLUMN1).int32(COLUMN2).build(BODY_SCHEMA),
                        new AvroSchemaBuilder(1).int64(FOOTER_FIELD1).build(FOOTER_SCHEMA)));

        final ByteArrayOutputStream out = new ByteArrayOutputStream();
        final DataFileWriter<GenericRecord> writer = new DataFileWriter<GenericRecord>(new GenericDatumWriter<GenericRecord>(schema)).create(schema, out);

        final AvroRecordBuilder headerBuilder = new AvroRecordBuilder(schema.getTypes().get(0));
        writer.append(headerBuilder.field(version).build());

        final AvroRecordBuilder bodyBuilder = new AvroRecordBuilder(schema.getTypes().get(1));
        writer.append(bodyBuilder.field(CITY1).field(POPULATION1).build());
        writer.append(bodyBuilder.field(CITY2).field(POPULATION2).build());

        final AvroRecordBuilder footerBuilder = new AvroRecordBuilder(schema.getTypes().get(2));
        writer.append(footerBuilder.field(total).build());

        writer.close();

To read it back we again assume we know what is in the input stream. I a real application one would need to compare schema names against a well-known/hard-coded list of alternatives (in this test, HEADER, DATA and FOOTER) to understand how to interpret a record. Actually, it would be a good idea to check in production code that the first record is always of type HEADER and the last of type FOOTER.

final DataFileStream<GenericRecord> reader = new DataFileStream<GenericRecord>(new ByteArrayInputStream(out.toByteArray()), new GenericDatumReader<GenericRecord>());
        GenericRecord deserialized = null;

        assertTrue(reader.hasNext());
        deserialized = reader.next(deserialized);
        assertEquals(HEADER_SCHEMA, deserialized.getSchema().getName());
        assertEquals(version, deserialized.get(HEADER_FIELD1));

        assertTrue(reader.hasNext());
        deserialized = reader.next(deserialized);
        assertEquals(BODY_SCHEMA, deserialized.getSchema().getName());
        assertEquals(new Utf8(CITY1), deserialized.get(COLUMN1));
        assertEquals(POPULATION1, deserialized.get(COLUMN2));

        assertTrue(reader.hasNext());
        deserialized = reader.next(deserialized);
        assertEquals(BODY_SCHEMA, deserialized.getSchema().getName());
        assertEquals(new Utf8(CITY2), deserialized.get(COLUMN1));
        assertEquals(POPULATION2, deserialized.get(COLUMN2));

        assertTrue(reader.hasNext());
        deserialized = reader.next(deserialized);
        assertEquals(FOOTER_SCHEMA, deserialized.getSchema().getName());
        assertEquals(total, deserialized.get(FOOTER_FIELD1));

        assertFalse(reader.hasNext());

In this case the generated schema looks slightly more interesting:

[ {
  "type" : "record",
  "name" : "HEADER",
  "fields" : [ {
    "name" : "VERSION",
    "type" : "int"
  } ]
}, {
  "type" : "record",
  "name" : "DATA",
  "fields" : [ {
    "name" : "CITY",
    "type" : "string"
  }, {
    "name" : "POPULATION",
    "type" : "int"
  } ]
}, {
  "type" : "record",
  "name" : "FOOTER",
  "fields" : [ {
    "name" : "CHKSUM",
    "type" : "long"
  } ]
} ]