flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Robert Metzger <rmetz...@apache.org>
Subject Re: Custom Class for state checkpointing
Date Tue, 18 Aug 2015 15:10:46 GMT
I'm still working on writing a test case for reproducing the issue.
Which Flink version are you using?
If you are using 0.10-SNAPSHOT, which exact commit?

On Tue, Aug 18, 2015 at 2:09 PM, Robert Metzger <rmetzger@apache.org> wrote:

> I created a JIRA for the issue:
> https://issues.apache.org/jira/browse/FLINK-2543
>
> Once I'm done with the Kafka pull request, I'll take a look into this.
>
> On Tue, Aug 18, 2015 at 1:56 PM, Stephan Ewen <sewen@apache.org> wrote:
>
>> Yep, that is a valid bug!
>> State is apparently not resolved with the correct classloader.
>>
>> As a workaround, you can checkpoint byte arrays and serialize/deserialize
>> the state into byte arrays yourself. You can use the apache commons
>> SerializationUtil class, or Flinks InstantiationUtil class for that.
>>
>> You can get the ClassLoader for the user code (needed for
>> deserialization) via "getRuntimeContext().getUserCodeClassLoader()".
>>
>> Let us know if that workaround works. We'll try to get a fix for that out
>> very soon!
>>
>> Greetings,
>> Stephan
>>
>>
>>
>> On Tue, Aug 18, 2015 at 12:23 PM, Robert Metzger <rmetzger@apache.org>
>> wrote:
>>
>>> Java's HashMap is serializable.
>>> If it is only the map, you can just use the HashMap<> as the state.
>>>
>>> If you have more data, you can use TupleX, for example:
>>>
>>> Tuple2<HashMap<Integer, String>, Long>(myMap, myLong);
>>>
>>>
>>> On Tue, Aug 18, 2015 at 12:21 PM, Rico Bergmann <info@ricobergmann.de>
>>> wrote:
>>>
>>>> Hi!
>>>>
>>>> Using TupleX is not possible since the state is very big (a Hashtable).
>>>>
>>>> How would I have to do serialization into a byte array?
>>>>
>>>> Greets. Rico.
>>>>
>>>>
>>>>
>>>> Am 18.08.2015 um 11:44 schrieb Robert Metzger <rmetzger@apache.org>:
>>>>
>>>> Hi Rico,
>>>>
>>>> I'm pretty sure that this is a valid bug you've found, since this case
>>>> is not yet tested (afaik).
>>>> We'll fix the issue asap, until then, are you able to encapsulate your
>>>> state in something that is available in Flink, for example a TupleX or just
>>>> serialize it yourself into a byte[] ?
>>>>
>>>> On Tue, Aug 18, 2015 at 11:32 AM, Rico Bergmann <info@ricobergmann.de>
>>>> wrote:
>>>>
>>>>> Hi!
>>>>> Is it possible to use your own class?
>>>>> I'm using the file state handler at the Jobmanager and implemented the
>>>>> Checkpointed interface.
>>>>>
>>>>> I tried this and got an exception:
>>>>>
>>>>> Error: java.lang.RuntimeException: Failed to deserialize state handle
>>>>> and setup initial operator state.
>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:544)
>>>>> at java.lang.Thread.run(Thread.java:745)
>>>>> Caused by: java.lang.ClassNotFoundException:
>>>>> com.ottogroup.bi.searchlab.searchsessionizer.OperatorState
>>>>>
>>>>> at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
>>>>> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
>>>>> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
>>>>> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
>>>>> at java.lang.Class.forName0(Native Method)
>>>>> at java.lang.Class.forName(Class.java:348)
>>>>> at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:626)
>>>>> at
>>>>> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1613)
>>>>> at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1518)
>>>>> at
>>>>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1774)
>>>>> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
>>>>> at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
>>>>> at
>>>>> org.apache.flink.runtime.state.ByteStreamStateHandle.getState(ByteStreamStateHandle.java:63)
>>>>> at
>>>>> org.apache.flink.runtime.state.ByteStreamStateHandle.getState(ByteStreamStateHandle.java:33)
>>>>> at
>>>>> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.restoreInitialState(AbstractUdfStreamOperator.java:83)
>>>>> at
>>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.setInitialState(StreamTask.java:276)
>>>>> at
>>>>> org.apache.flink.runtime.state.StateUtils.setOperatorState(StateUtils.java:51)
>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:541)
>>>>>
>>>>>
>>>>
>>>
>>
>

Mime
View raw message