主页 / coding / 问题 / 79252786
Accepted
Baiqing
Asked: 2024-12-05 05:29:07 +0800 CST

为 Flink 提供类型信息以便在没有 Kryo 的情况下序列化累加器对象

  • 772

我有一个自定义的 AggregateFunction，其签名如下：

public class CustomAggregateFunction
    implements AggregateFunction<CustomInput, AggregationAccumulator, CustomOutput> { code...}

MyAggregationAccumulator是包含一些地图的简单对象，带有 Lombok @Data 注释

  @Data
  public static class AggregationAccumulator {
    private Map<String, Long> customMap = new HashMap<>();
  }

然而，Flink 表示13:18:50,091 INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - Field AggregationAccumulator#customMap will be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance and schema evolution.

我如何提供类型信息以便它不会使用 Kryo？

java

1 个回答

  1. Best Answer

    @TypeInfo您可以在地图上添加注释，并提供适当的TypeInfoFactory实现。如下所示：

    import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.Map;

import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.typeinfo.TypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.MapTypeInfo;

public class AccumulatorWithMap {

    @TypeInfo(MapTypeInfoFactory.class)
    private Map<String, Long> customMap = new HashMap<>();
    
    public AccumulatorWithMap() { }

    public Map<String, Long> getCustomMap() {
        return customMap;
    }

    public void setCustomMap(Map<String, Long> customMap) {
        this.customMap = customMap;
    }
    
    public static class MapTypeInfoFactory<K, V> extends TypeInfoFactory<Map<K, V>> {

        @SuppressWarnings("unchecked")
        @Override
        public TypeInformation<Map<K, V>> createTypeInfo(
                Type t, Map<String, TypeInformation<?>> genericParameters) {
            TypeInformation<K> keyType = (TypeInformation<K>)genericParameters.get("K");
            TypeInformation<V> valueType = (TypeInformation<V>)genericParameters.get("V");

            if (keyType == null) {
                throw new InvalidTypesException(
                        "Type extraction is not possible on Map"
                                + " type as it does not contain information about the 'key' type.");
            }

            if (valueType == null) {
                throw new InvalidTypesException(
                        "Type extraction is not possible on Map"
                                + " type as it does not contain information about the 'value' type.");
            }
            
            return new MapTypeInfo<K, V>(keyType, valueType);
        }
    }

    
}

    为了验证，请使用：

    PojoTestUtils.assertSerializedAsPojoWithoutKryo(AccumulatorWithMap.class);
    • 1

相关问题