DataType

DataType：flink table的基本数据类型其子类继承关系:

图1DataType继承关系

其中将java的基本数据类型和基本数据类型的装箱数据类型统一划分原子类型。table中每一列column对应的数据类型InternalType.DataType没有绑定任何数据结构，序列化之前会转化为二进制的形式便于分布式计算

类型信息TypeInformation

在内部，Flink在类型之间分类：

基本类型：所有的Java基本类型和装箱类型以及void，String，Date，BigDecimal，和BigInteger。

基元数组和对象数组：

复合类型：

        Flink Java Tuples（Flink Java API的一部分）：最多25个字段，不支持空字段

        Scala Case Class（包括Scala元组）：最多22个字段，不支持空字段

        Row：具有任意数量字段的元组并支持空字段

        POJO：遵循某种类似bean的模式的类

辅助类型（选项，任一，列表，Map，......）

通用类型：这些不会被Flink本身序列化，而是由Kryo序列化。

类型信息用于flink函数输入输出数据的类型，flink看使用类型信息来生成序列化器和比较器,并提供语义检查。将代码中的model/object字段映射到schema的列中。映射的column字段数据类型和DataType类型对应。每一个TypeInformation都提供类型序列化器。（TypeInformation实现类较多不列举）

 public abstract TypeSerializer<T> createSerializer(ExecutionConfig ;

Table中的Column中数据类型是DataType,数据需要序列化的时候先是根据DataType转化对应的TypeInformation,然后创建对应得到序列化其器。序列化器从MemorySgment中拿数据。进行序列化操作。

1、DataType和TypeInformation之间相互转化

org.apache.flink.table.api.types.TypeConverters提供了DataType和Typeinformation之间的相互转化。

图2Typeinformation===》InternalType

createInternalTypeFromTypeInfo图2提供了部分截图用于根据Typeinformation信息匹配返回对应的DataType.

TypeInformationWrappedDataType继承之ExternalType==>DataType.其是包装类型信息的外部类型。也有可能是复合类型（row,pojo,tuple,caseclass）.构造函数如下；

图3

他会循环递归的调用creatInternalTypeFromTypeInfo.直到返回的数据类型是基本数据类型

图4,和图2是一个方法

BasicTypeInfor维护许多基本数据类型 其中之一：

public static final BasicTypeInfo<Boolean> BOOLEAN_TYPE_INFO = new BasicTypeInfo<>(Boolean.class, new Class<?>[]{}, BooleanSerializer.INSTANCE, BooleanComparator.class);

DataTypes:

public static final BooleanType BOOLEAN = BooleanType.INSTANCE;

public class BooleanType extends PrimitiveType {

public static final BooleanType INSTANCE = new BooleanType();

private BooleanType() {}}

2、TypeInformation和DataType之间相互转化

提供了两个方法createExternalTypeInfoFromDataType和createInternalTypeInfoFromDataType。

createExternalTypeInfoFromDataType方法截图。针对数组类型是有java基本数据类型和装箱的数据类型。

图5

针对复合类型会递归调用自身。

图6

3、自定义的类型信息.利用工厂来创建Typeinformation。在自定义的数据类型中利用注解TypeInfo,

注解的使用

4、数据类型的序列化器

每一个数据类型都有其对应序列化器,flink有自身的序列化器和kryo序列化器，

自身序列化器以BaseMapSerializer（父类TypeSerializer）为例看序列化操作。其构造器

图7

其序列化函数为：

序列化函数

先将数据类型转化为BinaryMap,从内存BinaryMap的segment片段中拿到数据对应的字节数组，利用DataOutputSerializer进行序列化，写入到一个buffer中。

反序列化函数：

反序列化函数

将DataInputDeserializer中buffer数据写入到字节数据中，然后写入到对应BinaryMap的segments中。

kryo序列化器和自定义序列化器的使用。

//强制使用kryo序列化器。finalExecutionEnvironmentenv=ExecutionEnvironment.getExecutionEnvironment();env.getConfig().enableForceKryo();

//注册子能够定义序列化器，env.getExecutionConfig().addDefaultKryoSerializer(TestPojo.class, (Class) CustomKryoTestSerializer.class);

自定义序列化器

JavaSerializer的序列化和反序列化过程。Kryo里面维护一个objectMap,key对应当前对象，value对应的是序列化是需要用到输入输出流。

序列化 反序列

Kryo序列化器KryoSerializer（父类TypeSerializer）：用于序列化不包含在基本类型、元组和pojos中。kryo.writeClassAndObject(output, record);output是对DataOutputView的包装，record需要序列化的数据，kryo.readClassAndObject(input)反序列化。