Apache Flink类型及序列化研读&生产应用｜得物技术

一

背景

二

简单理论阐述(基于Flink 1.13)

支持的数据类型

TypeInformation

何时需要数据类型获取

@Experimentalpublic <OUT> DataStreamSource<OUT> fromSource(        Source<OUT, ?, ?> source,        WatermarkStrategy<OUT> timestampsAndWatermarks,        String sourceName,        TypeInformation<OUT> typeInfo) {
    final TypeInformation<OUT> resolvedTypeInfo =            getTypeInfo(source, sourceName, Source.class, typeInfo);
    return new DataStreamSource<>(            this,            checkNotNull(source, "source"),            checkNotNull(timestampsAndWatermarks, "timestampsAndWatermarks"),            checkNotNull(resolvedTypeInfo),            checkNotNull(sourceName));}

/** Returns the {@code TypeInformation} for the elements of the input. */public TypeInformation<IN> getInputType() {    return input.getOutputType();}

@Overridepublic <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {    KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);    stateProperties.initializeSerializerUnlessSet(getExecutionConfig());    return keyedStateStore.getState(stateProperties);}
public void initializeSerializerUnlessSet(ExecutionConfig executionConfig) {    if (serializerAtomicReference.get() == null) {        checkState(typeInfo != null, "no serializer and no type info");        // try to instantiate and set the serializer        TypeSerializer<T> serializer = typeInfo.createSerializer(executionConfig);        // use cas to assure the singleton        if (!serializerAtomicReference.compareAndSet(null, serializer)) {            LOG.debug("Someone else beat us at initializing the serializer.");        }    }}

数据类型的自动推断

public <R> SingleOutputStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper) {    TypeInformation<R> outType = TypeExtractor.getFlatMapReturnTypes((FlatMapFunction)this.clean(flatMapper), this.getType(), Utils.getCallLocationName(), true);    return this.flatMap(flatMapper, outType);}

/** * A utility for reflection analysis on classes, to determine the return type of implementations of * transformation functions. * * <p>NOTES FOR USERS OF THIS CLASS: Automatic type extraction is a hacky business that depends on a * lot of variables such as generics, compiler, interfaces, etc. The type extraction fails regularly * with either {@link MissingTypeInfo} or hard exceptions. Whenever you use methods of this class, * make sure to provide a way to pass custom type information as a fallback. */

@PublicEvolvingpublic static <IN, OUT> TypeInformation<OUT> getUnaryOperatorReturnType(        Function function,        Class<?> baseClass,        int inputTypeArgumentIndex,        int outputTypeArgumentIndex,        int[] lambdaOutputTypeArgumentIndices,        TypeInformation<IN> inType,        String functionName,        boolean allowMissing) {
    Preconditions.checkArgument(            inType == null || inputTypeArgumentIndex >= 0,            "Input type argument index was not provided");    Preconditions.checkArgument(            outputTypeArgumentIndex >= 0, "Output type argument index was not provided");    Preconditions.checkArgument(            lambdaOutputTypeArgumentIndices != null,            "Indices for output type arguments within lambda not provided");
    // explicit result type has highest precedence    if (function instanceof ResultTypeQueryable) {        return ((ResultTypeQueryable<OUT>) function).getProducedType();    }
    // perform extraction    try {        final LambdaExecutable exec;        try {            exec = checkAndExtractLambda(function);        } catch (TypeExtractionException e) {            throw new InvalidTypesException("Internal error occurred.", e);        }        if (exec != null) {
            // parameters must be accessed from behind, since JVM can add additional parameters            // e.g. when using local variables inside lambda function            // paramLen is the total number of parameters of the provided lambda, it includes            // parameters added through closure            final int paramLen = exec.getParameterTypes().length;
            final Method sam = TypeExtractionUtils.getSingleAbstractMethod(baseClass);
            // number of parameters the SAM of implemented interface has; the parameter indexing            // applies to this range            final int baseParametersLen = sam.getParameterTypes().length;
            final Type output;            if (lambdaOutputTypeArgumentIndices.length > 0) {                output =                        TypeExtractionUtils.extractTypeFromLambda(                                baseClass,                                exec,                                lambdaOutputTypeArgumentIndices,                                paramLen,                                baseParametersLen);            } else {                output = exec.getReturnType();                TypeExtractionUtils.validateLambdaType(baseClass, output);            }
            return new TypeExtractor().privateCreateTypeInfo(output, inType, null);        } else {            if (inType != null) {                validateInputType(                        baseClass, function.getClass(), inputTypeArgumentIndex, inType);            }            return new TypeExtractor()                    .privateCreateTypeInfo(                            baseClass,                            function.getClass(),                            outputTypeArgumentIndex,                            inType,                            null);        }    } catch (InvalidTypesException e) {        if (allowMissing) {            return (TypeInformation<OUT>)                    new MissingTypeInfo(                            functionName != null ? functionName : function.toString(), e);        } else {            throw e;        }    }}

public class KafkaSource<OUT>        implements Source<OUT, KafkaPartitionSplit, KafkaSourceEnumState>,                ResultTypeQueryable<OUT>//deserializationSchema 是需要用户自己定义的。@Overridepublic TypeInformation<OUT> getProducedType() {    return deserializationSchema.getProducedType();}                //JSONKeyValueDeserializationSchema@Overridepublic TypeInformation<ObjectNode> getProducedType() {    return getForClass(ObjectNode.class);}

@Experimentalpublic <OUT> DataStreamSource<OUT> fromSource(Source<OUT, ?, ?> source, WatermarkStrategy<OUT> timestampsAndWatermarks, String sourceName, TypeInformation<OUT> typeInfo) {    TypeInformation<OUT> resolvedTypeInfo = this.getTypeInfo(source, sourceName, Source.class, typeInfo);    return new DataStreamSource(this, (Source)Preconditions.checkNotNull(source, "source"), (WatermarkStrategy)Preconditions.checkNotNull(timestampsAndWatermarks, "timestampsAndWatermarks"), (TypeInformation)Preconditions.checkNotNull(resolvedTypeInfo), (String)Preconditions.checkNotNull(sourceName));}

public <R> SingleOutputStreamOperator<R> map(        MapFunction<T, R> mapper, TypeInformation<R> outputType) {    return transform("Map", outputType, new StreamMap<>(clean(mapper)));}

@PublicEvolvingpublic <R> SingleOutputStreamOperator<R> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {    return this.doTransform(operatorName, outTypeInfo, SimpleOperatorFactory.of(operator));}

public <K> KeyedStream<T, K> keyBy(KeySelector<T, K> key, TypeInformation<K> keyType) {    Preconditions.checkNotNull(key);    Preconditions.checkNotNull(keyType);    return new KeyedStream(this, (KeySelector)this.clean(key), keyType);}

 public ValueStateDescriptor(String name, TypeInformation<T> typeInfo) {    super(name, typeInfo, (Object)null);}

自定义数据类型&自定义序列化器

@Datapublic class BroadcastConfig implements Serializable {    public String config_type;    public String date;    public String media_id;    public String account_id;    public String label_id;    public long start_time;    public long end_time;    public int interval;    public String msg;
    public BroadcastConfig() {    }    }

HashMap<String, TypeInformation<?>> pojoFieldName = new HashMap<>();pojoFieldName.put("config_type", Types.STRING);pojoFieldName.put("date", Types.STRING);pojoFieldName.put("media_id", Types.STRING);pojoFieldName.put("account_id", Types.STRING);pojoFieldName.put("label_id", Types.STRING);pojoFieldName.put("start_time", Types.LONG);pojoFieldName.put("end_time", Types.LONG);pojoFieldName.put("interval", Types.INT);pojoFieldName.put("msg", Types.STRING);
return Types.POJO(        BroadcastConfig.class,        pojoFieldName);

public  class BinaryRowDataTypeInfo extends TypeInformation<BinaryRowData> {
    private static final long serialVersionUID = 4786289562505208256L;    private final int numFields;    private final Class<BinaryRowData> clazz;    private final TypeSerializer<BinaryRowData> serializer;
    public BinaryRowDataTypeInfo(int numFields) {        this.numFields=numFields;        this.clazz=BinaryRowData.class;        serializer= new BinaryRowDataSerializer(numFields);    }
    @Override    public boolean isBasicType() {        return false;    }
    @Override    public boolean isTupleType() {        return false;    }
    @Override    public int getArity() {        return numFields;    }
    @Override    public int getTotalFields() {        return numFields;    }
    @Override    public Class<BinaryRowData> getTypeClass() {        return this.clazz;    }
    @Override    public boolean isKeyType() {        return false;    }
    @Override    public TypeSerializer<BinaryRowData> createSerializer(ExecutionConfig config) {        return serializer;    }
    @Override    public String toString() {        return "BinaryRowDataTypeInfo<" + clazz.getCanonicalName() + ">";    }
    @Override    public boolean equals(Object obj) {        if (obj instanceof BinaryRowDataTypeInfo) {
            BinaryRowDataTypeInfo that = (BinaryRowDataTypeInfo) obj;
            return that.canEqual(this)                    && this.numFields==that.numFields;        } else {            return false;        }    }
    @Override    public int hashCode() {        return Objects.hash(this.clazz,serializer.hashCode());    }
    @Override    public boolean canEqual(Object obj) {        return obj instanceof BinaryRowDataTypeInfo;    }}

public class Roaring64BitmapTypeSerializer extends TypeSerializer<Roaring64Bitmap> {    /**     * Sharable instance of the Roaring64BitmapTypeSerializer.     */    public static final Roaring64BitmapTypeSerializer INSTANCE = new Roaring64BitmapTypeSerializer();    private static final long serialVersionUID = -8544079063839253971L;
    @Override    public boolean isImmutableType() {        return false;    }
    @Override    public TypeSerializer<Roaring64Bitmap> duplicate() {        return this;    }
    @Override    public Roaring64Bitmap createInstance() {        return new Roaring64Bitmap();    }
    @Override    public Roaring64Bitmap copy(Roaring64Bitmap from) {        Roaring64Bitmap copiedMap = new Roaring64Bitmap();        from.forEach(copiedMap::addLong);        return copiedMap;    }
    @Override    public Roaring64Bitmap copy(Roaring64Bitmap from, Roaring64Bitmap reuse) {        from.forEach(reuse::addLong);        return reuse;    }
    @Override    public int getLength() {        return -1;    }
    @Override    public void serialize(Roaring64Bitmap record, DataOutputView target) throws IOException {        record.serialize(target);    }
    @Override    public Roaring64Bitmap deserialize(DataInputView source) throws IOException {        Roaring64Bitmap navigableMap = new Roaring64Bitmap();        navigableMap.deserialize(source);        return navigableMap;    }
    @Override    public Roaring64Bitmap deserialize(Roaring64Bitmap reuse, DataInputView source) throws IOException {        reuse.deserialize(source);        return reuse;    }
    @Override    public void copy(DataInputView source, DataOutputView target) throws IOException {        Roaring64Bitmap deserialize = this.deserialize(source);        copy(deserialize);    }
    @Override    public boolean equals(Object obj) {        if (obj == this) {            return true;        } else if (obj != null && obj.getClass() == Roaring64BitmapTypeSerializer.class) {            return true;        } else {            return false;        }    }
    @Override    public int hashCode() {        return this.getClass().hashCode();    }
    @Override    public TypeSerializerSnapshot<Roaring64Bitmap> snapshotConfiguration() {        return new Roaring64BitmapTypeSerializer.Roaring64BitmapSerializerSnapshot();    }
    public static final class Roaring64BitmapSerializerSnapshot            extends SimpleTypeSerializerSnapshot<Roaring64Bitmap> {
        public Roaring64BitmapSerializerSnapshot() {            super(() -> Roaring64BitmapTypeSerializer.INSTANCE);        }    }}

@Overridepublic Optional<RexNode> convert(CallExpression call, ConvertContext context) {    FunctionDefinition functionDefinition = call.getFunctionDefinition();
    // built-in functions without implementation are handled separately    if (functionDefinition instanceof BuiltInFunctionDefinition) {        final BuiltInFunctionDefinition builtInFunction =                (BuiltInFunctionDefinition) functionDefinition;        if (!builtInFunction.getRuntimeClass().isPresent()) {            return Optional.empty();        }    }
    TypeInference typeInference =            functionDefinition.getTypeInference(context.getDataTypeFactory());    if (typeInference.getOutputTypeStrategy() == TypeStrategies.MISSING) {        return Optional.empty();    }
    switch (functionDefinition.getKind()) {        case SCALAR:        case TABLE:            List<RexNode> args =                    call.getChildren().stream()                            .map(context::toRexNode)                            .collect(Collectors.toList());
            final BridgingSqlFunction sqlFunction =                    BridgingSqlFunction.of(                            context.getDataTypeFactory(),                            context.getTypeFactory(),                            SqlKind.OTHER_FUNCTION,                            call.getFunctionIdentifier().orElse(null),                            functionDefinition,                            typeInference);
            return Optional.of(context.getRelBuilder().call(sqlFunction, args));        default:            return Optional.empty();    }}

public static class LastValueWithRetractAccumulator {    public Object lastValue = null;    public Long lastOrder = null;    public List<Tuple2<Object, Long>> retractList = new ArrayList<>();    public MapView<Object, List<Long>> valueToOrderMap = new MapView<>();    public MapView<Long, List<Object>> orderToValueMap = new MapView<>();
    @Override    public boolean equals(Object o) {        if (this == o) {            return true;        }        if (!(o instanceof LastValueWithRetractAccumulator)) {            return false;        }        LastValueWithRetractAccumulator that = (LastValueWithRetractAccumulator) o;        return Objects.equals(lastValue, that.lastValue)                && Objects.equals(lastOrder, that.lastOrder)                && Objects.equals(retractList, that.retractList)                && valueToOrderMap.equals(that.valueToOrderMap)                && orderToValueMap.equals(that.orderToValueMap)                ;    }
    @Override    public int hashCode() {        return Objects.hash(lastValue, lastOrder, valueToOrderMap, orderToValueMap, retractList);    }
}

public TypeInference getTypeInference(DataTypeFactory typeFactory) {
    return TypeInference.newBuilder()            .accumulatorTypeStrategy(callContext -> {
                List<DataType> dataTypes = callContext.getArgumentDataTypes();
                DataType argDataType;                if (dataTypes.get(0)                        .getLogicalType()                        .getTypeRoot()                        .getFamilies()                        .contains(LogicalTypeFamily.CHARACTER_STRING)) {                    argDataType = DataTypes.STRING();                } else                    argDataType = DataTypeUtils.toInternalDataType(dataTypes.get(0));

                DataType accDataType = DataTypes.STRUCTURED(                        LastValueWithRetractAccumulator.class,                        DataTypes.FIELD("lastValue", argDataType.nullable()),                        DataTypes.FIELD("lastOrder", DataTypes.BIGINT()),                        DataTypes.FIELD("retractList", DataTypes.ARRAY(                                DataTypes.STRUCTURED(                                        Tuple2.class,                                        DataTypes.FIELD("f0", argDataType.nullable()),                                        DataTypes.FIELD("f1", DataTypes.BIGINT())                                )).bridgedTo(List.class)),                        DataTypes.FIELD(                                "valueToOrderMap",                                MapView.newMapViewDataType(                                        argDataType.nullable(),                                        DataTypes.ARRAY(DataTypes.BIGINT()).bridgedTo(List.class))),                        //todo:blink 使用SortedMapView 优化性能，开源使用MapView key天然字典升序，倒序遍历性能可能不佳                        DataTypes.FIELD(                                "orderToValueMap",                                MapView.newMapViewDataType(                                        DataTypes.BIGINT(),                                        DataTypes.ARRAY(argDataType.nullable()).bridgedTo(List.class)))                );
                return Optional.of(accDataType);            })            .build()            ;}

@Overridepublic TypeInference getTypeInference(DataTypeFactory typeFactory) {
    return TypeInference.newBuilder()            .outputTypeStrategy(callContext -> {
                List<DataType> dataTypes = callContext.getArgumentDataTypes();
                DataType argDataType;
                if (dataTypes.get(0)                        .getLogicalType()                        .getTypeRoot()                        .getFamilies()                        .contains(LogicalTypeFamily.CHARACTER_STRING)) {                    argDataType = DataTypes.STRING();                } else                    argDataType = DataTypeUtils.toInternalDataType(dataTypes.get(0));
                return Optional.of(argDataType);            })            .build()            ;}

.outputTypeStrategy(callContext -> {
    List<DataType> dataTypes = callContext.getArgumentDataTypes();
    DataType argDataType;
    if (dataTypes.get(0)            .getLogicalType()            .getTypeRoot()            .getFamilies()            .contains(LogicalTypeFamily.CHARACTER_STRING)) {        argDataType = DataTypes.STRING();    } else        argDataType = DataTypeUtils.toInternalDataType(dataTypes.get(0));
    return Optional.of(argDataType);}

自定义DataType

/** * Data type of an arbitrary serialized type. This type is a black box within the table * ecosystem and is only deserialized at the edges. * * <p>The raw type is an extension to the SQL standard. * * <p>This method assumes that a {@link TypeSerializer} instance is present. Use {@link * #RAW(Class)} for automatically generating a serializer. * * @param clazz originating value class * @param serializer type serializer * @see RawType */public static <T> DataType RAW(Class<T> clazz, TypeSerializer<T> serializer) {    return new AtomicDataType(new RawType<>(clazz, serializer));}

public TypeInference getTypeInference(DataTypeFactory typeFactory) {
    return TypeInference.newBuilder()            .accumulatorTypeStrategy(callContext -> {                DataType type = DataTypes.RAW(                        Roaring64Bitmap.class,                        Roaring64BitmapTypeSerializer.INSTANCE                );                return Optional.of(type);            })            .outputTypeStrategy(callContext -> Optional.of(DataTypes.BIGINT()))            .build()            ;}

四

结语