flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Stephan Ewen <se...@apache.org>
Subject Re: Serialization performance
Date Thu, 02 Mar 2017 20:06:53 GMT

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

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.


On Thu, Mar 2, 2017 at 8:01 PM, Newport, Billy <Billy.Newport@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?

View raw message