Hi! Thanks for this writeup, very cool stuff !
For part (1) - serialization: I think that can be made a bit nicer. Avro is a bit of an odd citizen in Flink, because Flink serialization is actually schema aware, but does not integrate with Avro. That's why Avro types go through Kryo. We should try and make Avro a first class citizen. - The first step is to have a proper AvroSerializer. We have implemented one already, see "org.apache.flink.api.java.typeutils.runtime.AvroSerializer". It works with the ReflectDatumReader/Writer, but would be a good base line for all types of avro-based serializers in Flink.. - Then we need to figure out how the Avro Serializer is instantiated. We could just let the "GenericTypeInfo" create an Avro serializer for Avro types, and Kryo for all else. - The change would eventually have to be behind a config flag in the ExecutionConfig (see "GenericTypeInfo.createSerializer()") to make sure we do not break the default serialization format within a major release version. A side note: If you actually use that through Beam, I am actually not sure what will happen, because as far as I know, Beam uses its completely own serialization system and Flink sees only byte coders from Beam. Aljoscha can probably add more detail here. For part (2) - the filters: If I understand correctly, you "split" the data into different result sets that go to different sinks? The DataStream API has a "split/select" type of construct which would help there, the DataSet API does not have something like that. If you look for peak performance, the demux output format seems like a good workaround on the DataSet side. Greetings, Stephan On Thu, Mar 2, 2017 at 8:01 PM, Newport, Billy <billy.newp...@gs.com> wrote: > We’ve been working on performance for the last while. We’re using flink > 1.2 right now. We are writing batch jobs which process avro and parquet > input files and produce parquet files. > > > > Flink serialization costs seem to be the most significant aspect of our > wall clock time. We have written a custom kryo serializer for GenericRecord > which is similar to the Spark one in that it reuses avro Datum reader and > writers but writes a hash fingerprint for the schema instead of the schema > itself. > > > > We have subclassed most of the Rich* classes in flink and now also pass to > the constructors a Builder class which has the avro schema. When flink > inflates these, the builders are inflated and preregister the avro schema > for the hash code in a static map which the inflation/writing code uses > later. > > > > This is working pretty well for us, it’s like 10-20x faster than just > using GenericRecords normally. The question is this: Is this the right way > to do this? If it is then we can contribute it and then how to modify beam > so that it uses this stuff under the covers, we can’t use beam at all right > now as far as I can tell because of the performance issues with > GenericRecord. > > > > The other performance optimization is basically removing filters which > again seem to double wall clock time. We wrote an embedded demux > outputformat which receives a Tuple<Enum,GenericRecord> and writes to a > different parquet file depending on Enum. This was 2x faster than a naïve 4 > filters going to 4 parquet outputformats. > > > > Do these optimizations seem unnecessary to some? Is there some trick we’re > missing? > > >
