flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Yukun Guo <gyk....@gmail.com>
Subject Re: Serialization problem for Guava's TreeMultimap
Date Wed, 21 Sep 2016 03:06:15 GMT
Thank you for quickly fixing it!


On 20 September 2016 at 17:17, Fabian Hueske <fhueske@gmail.com> wrote:

> Hi Yukun,
>
> I debugged this issue and found that this is a bug in the serialization of
> the StateDescriptor.
> I have created FLINK-4640 [1] to resolve the issue.
>
> Thanks for reporting the issue.
>
> Best, Fabian
>
> [1] https://issues.apache.org/jira/browse/FLINK-4640
>
> 2016-09-20 10:35 GMT+02:00 Yukun Guo <gyk.net@gmail.com>:
>
>> Some detail: if running the FoldFunction on a KeyedStream, everything
>> works fine. So it must relate to the way WindowedStream handles type
>> extraction.
>>
>> In case any Flink experts would like to reproduce it, I have created a
>> repo on Github: github.com/gyk/flink-multimap
>>
>> On 20 September 2016 at 10:33, Yukun Guo <gyk.net@gmail.com> wrote:
>>
>>> Hi,
>>>
>>> The same error occurs after changing the code, unfortunately.
>>>
>>> BTW, registerTypeWithKryoSerializer requires the 2nd argument to be a `T
>>> serializer` where T extends Serializer<?> & Serializable, so I pass
a
>>> custom GenericJavaSerializer<T>, but I guess this doesn't matter much.
>>>
>>>
>>> On 19 September 2016 at 18:02, Stephan Ewen <sewen@apache.org> wrote:
>>>
>>>> Hi!
>>>>
>>>> Can you use "env.getConfig().registerTypeW
>>>> ithKryoSerializer(TreeMultimap.class, JavaSerializer.class)" ?
>>>>
>>>> Best,
>>>> Stephan
>>>>
>>>>
>>>> On Sun, Sep 18, 2016 at 12:53 PM, Yukun Guo <gyk.net@gmail.com> wrote:
>>>>
>>>>> Here is the code snippet:
>>>>>
>>>>> windowedStream.fold(TreeMultimap.<Long, String>create(), new FoldFunction<Tuple2<String,
Long>, TreeMultimap<Long, String>>() {
>>>>>     @Override
>>>>>     public TreeMultimap<Long, String> fold(TreeMultimap<Long,
String> topKSoFar,
>>>>>                                            Tuple2<String, Long>
itemCount) throws Exception {
>>>>>         String item = itemCount.f0;
>>>>>         Long count = itemCount.f1;
>>>>>         topKSoFar.put(count, item);
>>>>>         if (topKSoFar.keySet().size() > topK) {
>>>>>             topKSoFar.removeAll(topKSoFar.keySet().first());
>>>>>         }
>>>>>         return topKSoFar;
>>>>>     }
>>>>> });
>>>>>
>>>>>
>>>>> The problem is when fold function getting called, the initial value
>>>>> has lost therefore it encounters a NullPointerException. This is due
to
>>>>> failed type extraction and serialization, as shown in the log message:
>>>>> "INFO  TypeExtractor:1685 - No fields detected for class
>>>>> com.google.common.collect.TreeMultimap. Cannot be used as a PojoType.
>>>>> Will be handled as GenericType."
>>>>>
>>>>> I have tried the following two ways to fix it but neither of them
>>>>> worked:
>>>>>
>>>>> 1. Writing a class TreeMultimapSerializer which extends Kryo's
>>>>> Serializer<T>, and calling `env.addDefaultKryoSerializer(TreeMultimap.class,
>>>>> new TreeMultimapSerializer()`. The write/read methods are almost
>>>>> line-by-line translations from TreeMultimap's own implementation.
>>>>>
>>>>> 2. TreeMultimap has implemented Serializable interface so Kryo can
>>>>> fall back to use the standard Java serialization. Since Kryo's
>>>>> JavaSerializer itself is not serializable, I wrote an adapter to make
it
>>>>> fit the "addDefaultKryoSerializer" API.
>>>>>
>>>>> Could you please give me some working examples for custom Kryo
>>>>> serialization in Flink?
>>>>>
>>>>>
>>>>> Best regards,
>>>>> Yukun
>>>>>
>>>>>
>>>>
>>>
>>
>

Mime
View raw message