DataType
DataType：flink table的基本数据类型其子类继承关系:
其中将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之间的相互转化。
createInternalTypeFromTypeInfo图2提供了部分截图用于根据Typeinformation信息匹配返回对应的DataType.
TypeInformationWrappedDataType继承之ExternalType==>DataType.其是包装类型信息的外部类型。也有可能是复合类型（row,pojo,tuple,caseclass）.构造函数如下；
他会循环递归的调用creatInternalTypeFromTypeInfo.直到返回的数据类型是基本数据类型
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基本数据类型和装箱的数据类型。
针对复合类型会递归调用自身。
3、自定义的类型信息.利用工厂来创建Typeinformation。在自定义的数据类型中利用注解TypeInfo,
4、数据类型的序列化器
每一个数据类型都有其对应序列化器,flink有自身的序列化器和kryo序列化器，
自身序列化器以BaseMapSerializer（父类TypeSerializer）为例看序列化操作。其构造器
其序列化函数为：
先将数据类型转化为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)反序列化。
