title: How can i use custom serialize method in Flink?
body: <p>I use <code>AtomicLongMap</code> in the construct function of RichMapFunctin.Like</p>
  
  <pre><code>public class PathAnalysis extends RichMapFunction&lt;ApiLog, ApiLog&gt; {
  
  private final AtomicLongMap&lt;Object&gt; mObjectAtomicLongMap;
  
  public PathAnalysis()
  {
      mObjectAtomicLongMap = AtomicLongMap.create();
  }
  }
  </code></pre>
  
  <p>register the custom serizlize class, but it not work        </p>
  
  <pre><code>env.getConfig().registerTypeWithKryoSerializer(AtomicLongMap.class, new AtomicLongMapSerializer());
  </code></pre>
  
  <p>It cause:</p>
  
  <pre><code>org.apache.flink.api.common.InvalidProgramException: The implementation of the RichMapFunction is not serializable. The object probably contains or references non serializable fields.
      at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:99)
      at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1550)
      at org.apache.flink.streaming.api.datastream.DataStream.clean(DataStream.java:184)
      at org.apache.flink.streaming.api.datastream.DataStream.map(DataStream.java:528)
      at com.ghzs.Topology.main(Topology.java:91)
  Caused by: java.io.NotSerializableException: org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.util.concurrent.AtomicLongMap
      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.writeObject(ObjectOutputStream.java:348)
      at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:315)
      at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:81)
      ... 4 more
  </code></pre>
  
  <p><code>AtomicLongMap</code> is not implements of <code>Serializable</code> .How can I register a effective custom serialize method?</p>