Re: Using guava ImmutableMap Serializer from magro/kyro-serializers
Thanks JingsongLee. You are right. I have registered all the implementation class and it worked. -- Sent from: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/
Re: Using guava ImmutableMap Serializer from magro/kyro-serializers
Hi @Vijay Srinivasaraghavan, as far as I know, kryo need register all
implementation classes instead of abstract class or interface.
You can see the detail in
de.javakaffee.kryoserializers.guava.ImmutableMapSerializer.registerSerializers.
Maybe you can try do same thing to ExecutionConfig like
ImmutableMapSerializer.registerSerializers:
private enum DummyEnum {
VALUE1,
VALUE2
}
private static void registerSerializers(ExecutionConfig conf) {
Class serializer = ImmutableMapSerializer.class;
conf.registerTypeWithKryoSerializer(ImmutableMap.class, serializer);
conf.registerTypeWithKryoSerializer(ImmutableMap.of().getClass(),
serializer);
Object o1 = new Object();
Object o2 = new Object();
conf.registerTypeWithKryoSerializer(ImmutableMap.of(o1, o1).getClass(),
serializer);
conf.registerTypeWithKryoSerializer(ImmutableMap.of(o1, o1, o2,
o2).getClass(), serializer);
Map enumMap = new EnumMap<>(DummyEnum.class);
for (DummyEnum e : DummyEnum.values()) {
enumMap.put(e, o1);
}
conf.registerTypeWithKryoSerializer(ImmutableMap.copyOf(enumMap).getClass(),
serializer);
ImmutableTable denseImmutableTable =
ImmutableTable.builder().put("a", 1, 1).put("b", 1, 1).build();
conf.registerTypeWithKryoSerializer(denseImmutableTable.rowMap().getClass(),
serializer);
conf.registerTypeWithKryoSerializer(((Map)denseImmutableTable.rowMap().get("a")).getClass(),
serializer);
conf.registerTypeWithKryoSerializer(denseImmutableTable.columnMap().getClass(),
serializer);
conf.registerTypeWithKryoSerializer(((Map)denseImmutableTable.columnMap().get(1)).getClass(),
serializer);
}
Best, JingsongLee
--
From:vijikarthi
Send Time:2019年4月17日(星期三) 04:45
To:dev
Subject:Using guava ImmutableMap Serializer from magro/kyro-serializers
Hello,
I am using Flink 1.7.2. I wrote a small application which uses
KeyedProcessFunction to maintain application state. The state value object
is using Guava's (version 18.0.50) ImmutableMap to create a copy of the map
instance as part of its constructor.
I am using "de.javakaffee:kryo-serializers:0.45" to take handle the guava
ImmutableMap serialization and configured the serializer using `
env.getConfig().registerTypeWithKryoSerializer(ImmutableMap.class,
ImmutableMapSerializer.class)`.
I am still seeing an issue with ImmutableMap serialization and I am not sure
if it is using the serializer that I have configured? (may be I am not
configuring it correctly??). Using standard java Map works properly.
Appreciate any inputs? I have provided the error stack trace and the code
snippet below.
Regards
Vijay
Error StackTrace
```
Serialization trace:
itemMap (com.flink.SensorState)
start (com.flink.ValueObject)
at
com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:133)
~[kryo-5.0.0-RC1.jar:na]
at
com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:122)
~[kryo-5.0.0-RC1.jar:na]
at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:712)
~[kryo-5.0.0-RC1.jar:na]
at
com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:114)
~[kryo-5.0.0-RC1.jar:na]
at
com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:122)
~[kryo-5.0.0-RC1.jar:na]
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:793)
~[kryo-5.0.0-RC1.jar:na]
at
org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:315)
~[flink-core-1.7.2.jar:1.7.2]
at
org.apache.flink.runtime.state.heap.StateTableByKeyGroupReaders.lambda$createV2PlusReader$0(StateTableByKeyGroupReaders.java:74)
~[flink-runtime_2.12-1.7.2.jar:1.7.2]
at
org.apache.flink.runtime.state.KeyGroupPartitioner$PartitioningResultKeyGroupReader.readMappingsInKeyGroup(KeyGroupPartitioner.java:297)
~[flink-runtime_2.12-1.7.2.jar:1.7.2]
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readKeyGroupStateData(HeapKeyedStateBackend.java:492)
~[flink-runtime_2.12-1.7.2.jar:1.7.2]
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readStateHandleStateData(HeapKeyedStateBackend.java:453)
~[flink-runtime_2.12-1.7.2.jar:1.7.2]
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restorePartitionedState(HeapKeyedStateBackend.java:410)
~[flink-runtime_2.12-1.7.2.jar:1.7.2]
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:358)
~[flink-runtime_2.12-1.7.2.jar:1.7.2]
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:104)
~[flink-runtime_2.12-1.7.2.jar:1.7.2]
at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:151)
~[flink-streaming-java_2.12-1.7.2.jar:1.7.2]
at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:123)
~[flink-streaming-java_2.12-1.7.2.jar:1.7.2]
at
Using guava ImmutableMap Serializer from magro/kyro-serializers
Hello, I am using Flink 1.7.2. I wrote a small application which uses KeyedProcessFunction to maintain application state. The state value object is using Guava's (version 18.0.50) ImmutableMap to create a copy of the map instance as part of its constructor. I am using "de.javakaffee:kryo-serializers:0.45" to take handle the guava ImmutableMap serialization and configured the serializer using ` env.getConfig().registerTypeWithKryoSerializer(ImmutableMap.class, ImmutableMapSerializer.class)`. I am still seeing an issue with ImmutableMap serialization and I am not sure if it is using the serializer that I have configured? (may be I am not configuring it correctly??). Using standard java Map works properly. Appreciate any inputs? I have provided the error stack trace and the code snippet below. Regards Vijay Error StackTrace ``` Serialization trace: itemMap (com.flink.SensorState) start (com.flink.ValueObject) at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:133) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:122) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:712) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:114) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:122) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:793) ~[kryo-5.0.0-RC1.jar:na] at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:315) ~[flink-core-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.StateTableByKeyGroupReaders.lambda$createV2PlusReader$0(StateTableByKeyGroupReaders.java:74) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.KeyGroupPartitioner$PartitioningResultKeyGroupReader.readMappingsInKeyGroup(KeyGroupPartitioner.java:297) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readKeyGroupStateData(HeapKeyedStateBackend.java:492) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readStateHandleStateData(HeapKeyedStateBackend.java:453) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restorePartitionedState(HeapKeyedStateBackend.java:410) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:358) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:104) ~[flink-runtime_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:151) ~[flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:123) ~[flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:284) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:135) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:250) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:738) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:289) [flink-streaming-java_2.12-1.7.2.jar:1.7.2] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704) [flink-runtime_2.12-1.7.2.jar:1.7.2] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_191] Caused by: java.lang.UnsupportedOperationException: null at org.apache.flink.shaded.guava18.com.google.common.collect.ImmutableMap.put(ImmutableMap.java:326) ~[flink-shaded-guava-18.0-5.0.jar:18.0-5.0] at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:230) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:42) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:712) ~[kryo-5.0.0-RC1.jar:na] at com.esotericsoftware.kryo.serializers.ReflectField.read(ReflectField.java:114) ~[kryo-5.0.0-RC1.jar:na] ... 22 common frames omitted ``` ``` final StreamExecutionEnvironment env =
