Posting an update here, because it came up again: Have a look at https://issues.apache.org/jira/browse/FLINK-17192 specifically this comment:
> There is a hidden/experimental feature in the sorter to offload large records, but it is not active by default. > > You can try and add "taskmanager.runtime.large-record-handler: true" to the config. > > The reason why it is a "hidden feature" is that it has some restrictions: The key must be serializable by Flink's default serializers and > regonized by the TypeExtractor, meaning you cannot use a custom serializer or a specific type information. > > For keys that are string, int, long, arrays, simple POJOs etc. it should be fine. For keys that are Avro types with a specific schema, > or types with custom serializers (including custom Kryo serializers) it might not work.