devbox
空白诗007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

从一例状态引发的性能问题谈Flink状态序列化

作者：空白诗007 | 2024-06-28 23:09:27

赞

踩

flink serialize map

前言

好久不见(鞠躬

最近处在转型期，每天忙到飞起，关注具体技术细节的精力自然就比较少了(上一篇许下的周更承诺也食言了 = =)。上周帮助他人快速解决了一个因误用Flink状态类型引发的性能问题，在这里做个quick notes，并简要介绍一下Flink状态序列化方面的基础知识。

问题及排查

上游部门同事反馈，一个计算逻辑并不复杂的多流join DataStream API作业频繁发生消费积压、checkpoint失败(现场截图已丢失)。作业拓扑如下图所示。

为了脱敏所以缩得很小 = =

按大状态作业的pattern对集群参数进行调优，未果。

通过Flink Web UI定位到问题点位于拓扑中倒数第二个算子，部分sub-task checkpoint总是过不去。观察Metrics面板，发现有少量数据倾斜，而上下游反压度量值全部为0。

经过持续观察，存在倾斜的sub-task数据量最多只比其他sub-task多出10%~15%，按照常理不应引起如此严重的性能问题。遂找到对应的TaskManager pod打印火焰图，结果如下。

可见RocksDB状态读写的耗时极长，大部分时间花在了Kryo序列化上，说明状态内存储了Flink序列化框架原生不支持的对象。直接让相关研发同学show me the code，真相大白：

  1. private transient MapState<String, HashSet<String>> state1;
  2. private transient MapState<String, HashSet<String>> state2;
  3. private transient ValueState<Map<String, String>> state3;

Flink序列化框架内并没有针对HashSet的序列化器，自然会fallback到Kryo。即使这些Set并不算大，状态操作的开销也会急剧上升。当然，ValueState<Map<String, String>>用法也是错误的，应改成MapState<String, String>。

最快的临时解决方法很简单：把所有状态内用到的HashSet全部改成Map<String, Boolean>，同样可以去重。虽然并不优雅，但因为有了原生MapSerializer支持，效率大幅提升。下面简要介绍Flink的状态序列化。

TypeSerializer

在我们创建状态句柄所需的描述符StateDescriptor时，要指定状态数据的类型，如：

  1. ValueStateDescriptor<Integer> stateDesc = new ValueStateDescriptor<>("myState", Integer.class);
  2. ValueState<Integer> state = this.getRuntimeContext().getState(stateDesc);

与此同时，也就指定了对应数据类型的Serializer。我们知道，TypeSerializer是Flink Runtime序列化机制的底层抽象，状态数据的序列化也不例外。以处理Map类型的MapSerializer为例，代码如下，比较清晰。

  1. @Internal
  2. public final class MapSerializer<K, V> extends TypeSerializer<Map<K, V>> {
  3.  
  4.     private static final long serialVersionUID = -6885593032367050078L;
  5.  
  6.     /** The serializer for the keys in the map */
  7.     private final TypeSerializer<K> keySerializer;
  8.  
  9.     /** The serializer for the values in the map */
  10.     private final TypeSerializer<V> valueSerializer;
  11.  
  12.     /**
  13.      * Creates a map serializer that uses the given serializers to serialize the key-value pairs in
  14.      * the map.
  15.      *
  16.      * @param keySerializer The serializer for the keys in the map
  17.      * @param valueSerializer The serializer for the values in the map
  18.      */
  19.     public MapSerializer(TypeSerializer<K> keySerializer, TypeSerializer<V> valueSerializer) {
  20.         this.keySerializer =
  21.                 Preconditions.checkNotNull(keySerializer, "The key serializer cannot be null");
  22.         this.valueSerializer =
  23.                 Preconditions.checkNotNull(valueSerializer, "The value serializer cannot be null.");
  24.     }
  25.  
  26.     // ------------------------------------------------------------------------
  27.     //  MapSerializer specific properties
  28.     // ------------------------------------------------------------------------
  29.  
  30.     public TypeSerializer<K> getKeySerializer() {
  31.         return keySerializer;
  32.     }
  33.  
  34.     public TypeSerializer<V> getValueSerializer() {
  35.         return valueSerializer;
  36.     }
  37.  
  38.     // ------------------------------------------------------------------------
  39.     //  Type Serializer implementation
  40.     // ------------------------------------------------------------------------
  41.  
  42.     @Override
  43.     public boolean isImmutableType() {
  44.         return false;
  45.     }
  46.  
  47.     @Override
  48.     public TypeSerializer<Map<K, V>> duplicate() {
  49.         TypeSerializer<K> duplicateKeySerializer = keySerializer.duplicate();
  50.         TypeSerializer<V> duplicateValueSerializer = valueSerializer.duplicate();
  51.  
  52.         return (duplicateKeySerializer == keySerializer)
  53.                         && (duplicateValueSerializer == valueSerializer)
  54.                 ? this
  55.                 : new MapSerializer<>(duplicateKeySerializer, duplicateValueSerializer);
  56.     }
  57.  
  58.     @Override
  59.     public Map<K, V> createInstance() {
  60.         return new HashMap<>();
  61.     }
  62.  
  63.     @Override
  64.     public Map<K, V> copy(Map<K, V> from) {
  65.         Map<K, V> newMap = new HashMap<>(from.size());
  66.  
  67.         for (Map.Entry<K, V> entry : from.entrySet()) {
  68.             K newKey = keySerializer.copy(entry.getKey());
  69.             V newValue = entry.getValue() == null ? null : valueSerializer.copy(entry.getValue());
  70.  
  71.             newMap.put(newKey, newValue);
  72.         }
  73.  
  74.         return newMap;
  75.     }
  76.  
  77.     @Override
  78.     public Map<K, V> copy(Map<K, V> from, Map<K, V> reuse) {
  79.         return copy(from);
  80.     }
  81.  
  82.     @Override
  83.     public int getLength() {
  84.         return -1; // var length
  85.     }
  86.  
  87.     @Override
  88.     public void serialize(Map<K, V> map, DataOutputView target) throws IOException {
  89.         final int size = map.size();
  90.         target.writeInt(size);
  91.  
  92.         for (Map.Entry<K, V> entry : map.entrySet()) {
  93.             keySerializer.serialize(entry.getKey(), target);
  94.  
  95.             if (entry.getValue() == null) {
  96.                 target.writeBoolean(true);
  97.             } else {
  98.                 target.writeBoolean(false);
  99.                 valueSerializer.serialize(entry.getValue(), target);
  100.             }
  101.         }
  102.     }
  103.  
  104.     @Override
  105.     public Map<K, V> deserialize(DataInputView source) throws IOException {
  106.         final int size = source.readInt();
  107.  
  108.         final Map<K, V> map = new HashMap<>(size);
  109.         for (int i = 0; i < size; ++i) {
  110.             K key = keySerializer.deserialize(source);
  111.  
  112.             boolean isNull = source.readBoolean();
  113.             V value = isNull ? null : valueSerializer.deserialize(source);
  114.  
  115.             map.put(key, value);
  116.         }
  117.  
  118.         return map;
  119.     }
  120.  
  121.     @Override
  122.     public Map<K, V> deserialize(Map<K, V> reuse, DataInputView source) throws IOException {
  123.         return deserialize(source);
  124.     }
  125.  
  126.     @Override
  127.     public void copy(DataInputView source, DataOutputView target) throws IOException {
  128.         final int size = source.readInt();
  129.         target.writeInt(size);
  130.  
  131.         for (int i = 0; i < size; ++i) {
  132.             keySerializer.copy(source, target);
  133.  
  134.             boolean isNull = source.readBoolean();
  135.             target.writeBoolean(isNull);
  136.  
  137.             if (!isNull) {
  138.                 valueSerializer.copy(source, target);
  139.             }
  140.         }
  141.     }
  142.  
  143.     @Override
  144.     public boolean equals(Object obj) {
  145.         return obj == this
  146.                 || (obj != null
  147.                         && obj.getClass() == getClass()
  148.                         && keySerializer.equals(((MapSerializer<?, ?>) obj).getKeySerializer())
  149.                         && valueSerializer.equals(
  150.                                 ((MapSerializer<?, ?>) obj).getValueSerializer()));
  151.     }
  152.  
  153.     @Override
  154.     public int hashCode() {
  155.         return keySerializer.hashCode() * 31 + valueSerializer.hashCode();
  156.     }
  157.  
  158.     // --------------------------------------------------------------------------------------------
  159.     // Serializer configuration snapshotting
  160.     // --------------------------------------------------------------------------------------------
  161.  
  162.     @Override
  163.     public TypeSerializerSnapshot<Map<K, V>> snapshotConfiguration() {
  164.         return new MapSerializerSnapshot<>(this);
  165.     }
  166. }

总结：

  • 序列化和反序列化本质上都是对MemorySegment的操作，通过DataOutputView写出二进制数据，通过DataInputView读入二进制数据；
  • 对于复合数据类型，也应嵌套定义并调用内部元素类型的TypeSerializer；
  • 必须要有对应的TypeSerializerSnapshot。该组件定义了TypeSerializer本身及其所包含的元数据(即state schema)的序列化方式，这些信息会存储在快照中。可见，通过TypeSerializerSnapshot可以判断状态恢复时数据的兼容性，是Flink实现state schema evolution特性的关键所在。

TypeSerializerSnapshot

TypeSerializerSnapshot接口有以下几个重要的方法。注释写得很清晰，不再废话了(实际是因为懒而且累 = =

  1. /**
  2.      * Returns the version of the current snapshot's written binary format.
  3.      *
  4.      * @return the version of the current snapshot's written binary format.
  5.      */
  6.     int getCurrentVersion();
  7.  
  8.     /**
  9.      * Writes the serializer snapshot to the provided {@link DataOutputView}. The current version of
  10.      * the written serializer snapshot's binary format is specified by the {@link
  11.      * #getCurrentVersion()} method.
  12.      *
  13.      * @param out the {@link DataOutputView} to write the snapshot to.
  14.      * @throws IOException Thrown if the snapshot data could not be written.
  15.      * @see #writeVersionedSnapshot(DataOutputView, TypeSerializerSnapshot)
  16.      */
  17.     void writeSnapshot(DataOutputView out) throws IOException;
  18.  
  19.     /**
  20.      * Reads the serializer snapshot from the provided {@link DataInputView}. The version of the
  21.      * binary format that the serializer snapshot was written with is provided. This version can be
  22.      * used to determine how the serializer snapshot should be read.
  23.      *
  24.      * @param readVersion version of the serializer snapshot's written binary format
  25.      * @param in the {@link DataInputView} to read the snapshot from.
  26.      * @param userCodeClassLoader the user code classloader
  27.      * @throws IOException Thrown if the snapshot data could be read or parsed.
  28.      * @see #readVersionedSnapshot(DataInputView, ClassLoader)
  29.      */
  30.     void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader)
  31.             throws IOException;
  32.  
  33.     /**
  34.      * Recreates a serializer instance from this snapshot. The returned serializer can be safely
  35.      * used to read data written by the prior serializer (i.e., the serializer that created this
  36.      * snapshot).
  37.      *
  38.      * @return a serializer instance restored from this serializer snapshot.
  39.      */
  40.     TypeSerializer<T> restoreSerializer();
  41.  
  42.     /**
  43.      * Checks a new serializer's compatibility to read data written by the prior serializer.
  44.      *
  45.      * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
  46.      * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
  47.      * used by the program that restores the checkpoint/savepoint. The outcome can be that the
  48.      * serialization format is compatible, that the program's serializer needs to reconfigure itself
  49.      * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
  50.      * that the format is outright incompatible, or that a migration needed. In the latter case, the
  51.      * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
  52.      * program's serializer re-serializes the data, thus converting the format during the restore
  53.      * operation.
  54.      *
  55.      * @param newSerializer the new serializer to check.
  56.      * @return the serializer compatibility result.
  57.      */
  58.     TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
  59.             TypeSerializer<T> newSerializer);

特别注意，在状态恢复时，state schema的兼容性判断结果TypeSerializerSchemaCompatibility有4种：

  • COMPATIBLE_AS_IS：兼容，可以直接使用新Serializer；
  • COMPATIBLE_AFTER_MIGRATION：兼容，但需要用快照中的旧Serializer反序列化一遍数据，再将数据用新Serializer重新序列化。最常见的场景如状态POJO中增加或删除字段，详情可以参考PojoSerializerSnapshot类的相关代码；
  • COMPATIBLE_WITH_RECONFIGURED_SERIALIZER：兼容，但需要将新Serializer重新配置之后再使用。此类场景不太常见，举例如状态POJO的类继承关系发生变化；
  • INCOMPATIBLE：不兼容，无法恢复。例如，更改POJO中的一个简单类型字段的type(e.g. String → Integer)，由于负责处理简单数据类型的SimpleTypeSerializerSnapshot不支持此类更改，就会抛出异常：
  1. @Override
  2.     public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
  3.             TypeSerializer<T> newSerializer) {
  4.  
  5.         return newSerializer.getClass() == serializerSupplier.get().getClass()
  6.                 ? TypeSerializerSchemaCompatibility.compatibleAsIs()
  7.                 : TypeSerializerSchemaCompatibility.incompatible();
  8.     }

显然，对于复合类型(如List、Map)，需要先判断外部容器Serializer的兼容性，再判断嵌套Serializer的兼容性。详情可以参考Flink内部专门为此定义的CompositeTypeSerializerSnapshot抽象类，该类比较复杂，在此按下不表。

The End

在一些特殊的场景下，我们需要自定义Serializers来实现更好的状态序列化(例如用RoaringBitmap代替Set在状态中进行高效的去重)，今天时间已经很晚，暂时不给出具体实现了。关于自定义状态序列化器的更多细节，请看官参见官方文档<<Custom Serialization for Managed State>>一章。

晚安晚安。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/空白诗007/article/detail/767622
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号