com.esotericsoftware.kryo.KryoException: Failed to serialize object.
	at org.apache.flink.streaming.python.util.serialization.PyObjectSerializer.write(PyObjectSerializer.java:41)
	at org.apache.flink.streaming.python.util.serialization.PyObjectSerializer.write(PyObjectSerializer.java:33)
	at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:505)
	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:224)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:577)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:689)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:667)
	at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)
	at org.apache.flink.streaming.python.util.PythonCollector.collect(PythonCollector.java:41)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.python.core.PyReflectedFunction.__call__(PyReflectedFunction.java:186)
	at org.python.core.PyReflectedFunction.__call__(PyReflectedFunction.java:204)
	at org.python.core.PyObject.__call__(PyObject.java:478)
	at org.python.core.PyObject.__call__(PyObject.java:482)
	at org.python.core.PyMethod.__call__(PyMethod.java:141)
	at GreatReviewsSpoltPort$py.flatMap$2(/var/folders/t1/gcltcjcn5zdgqfqrc32xk90x85xkg9/T/flink-dist-cache-4067f255-99c5-4d4f-8d22-38886386d7f9/31de3f085d427909d27ec63ca7ecd5d4/flink_streaming_plan_f17070b6-8f88-4284-8533-c3201b3fbbb2/GreatReviewsSpoltPort.py:125)
	at GreatReviewsSpoltPort$py.call_function(/var/folders/t1/gcltcjcn5zdgqfqrc32xk90x85xkg9/T/flink-dist-cache-4067f255-99c5-4d4f-8d22-38886386d7f9/31de3f085d427909d27ec63ca7ecd5d4/flink_streaming_plan_f17070b6-8f88-4284-8533-c3201b3fbbb2/GreatReviewsSpoltPort.py)
	at org.python.core.PyTableCode.call(PyTableCode.java:167)
	at org.python.core.PyBaseCode.call(PyBaseCode.java:307)
	at org.python.core.PyBaseCode.call(PyBaseCode.java:198)
	at org.python.core.PyFunction.__call__(PyFunction.java:482)
	at org.python.core.PyMethod.instancemethod___call__(PyMethod.java:237)
	at org.python.core.PyMethod.__call__(PyMethod.java:228)
	at org.python.core.PyMethod.__call__(PyMethod.java:218)
	at org.python.core.PyMethod.__call__(PyMethod.java:213)
	at org.python.core.PyObject._jcallexc(PyObject.java:3626)
	at org.python.proxies.GreatReviewsSpoltPort$GreatReviewsSpoltPort$0.flatMap(Unknown Source)
	at org.apache.flink.streaming.python.api.functions.PythonFlatMapFunction.flatMap(PythonFlatMapFunction.java:59)
	at org.apache.flink.streaming.python.api.functions.PythonFlatMapFunction.flatMap(PythonFlatMapFunction.java:40)
	at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:50)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:689)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:667)
	at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:689)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:667)
	at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
	at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111)
	at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordWithTimestamp(AbstractFetcher.java:398)
	at org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher.emitRecord(Kafka010Fetcher.java:89)
	at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:154)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:738)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.NotSerializableException: org.apache.avro.generic.GenericData$Record
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
	at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
	at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
	at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
	at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
	at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
	at org.apache.flink.streaming.python.util.serialization.SerializationUtils.serializeObject(SerializationUtils.java:38)
	at org.apache.flink.streaming.python.util.serialization.PyObjectSerializer.write(PyObjectSerializer.java:37)
	... 57 more