前言
这一讲将介绍一下序列化机制和过程函数(processfunction)。
序列化机制
使用 Flink 编写处理逻辑时,新手总是容易被林林总总的概念所混淆:
为什么 Flink 有那么多的类型声明方式?BasicTypeInfo.STRING_TYPE_INFO、Types.STRING、Types.STRING() 有何区别?TypeInfoFactory 又是什么?TypeInformation.of 和 TypeHint 是如何使用的呢?
接下来本文将逐步解密 Flink 的类型和序列化机制 (TypeInformation)。
Flink 的类型系统源码位于 org.apache.flink.api.common.typeinfo 包,让我们对上图 TypeInformation 深入追踪,看一下类的继承关系图:
可以看到,上面两个图片是一一对应的,TypeInformation 类是描述一切类型的公共基类,它和它的所有子类必须可序列化(Serializable),因为类型信息将会伴随 Flink 的作业提交,被传递给每个执行节点。
由于 Flink 自己管理内存,采用了一种非常紧凑的存储格式(见官方博文),因而类型信息在整个数据处理流程中属于至关重要的元数据。
- TypeExtractror 类型提取
Flink 内部实现了名为 TypeExtractror 的类,可以利用方法签名、子类信息等蛛丝马迹,自动提取和恢复类型信息(当然也可以显式声明,即本文所介绍的内容)。
然而由于 Java 的类型擦除,自动提取并不是总是有效。因而一些情况下(例如通过 URLClassLoader 动态加载的类),仍需手动处理;例如下图中对 DataSet 变换时,使用 .returns() 方法声明返回类型。
这里需要说明一下,returns() 接受三种类型的参数:字符串描述的类名(例如 “String”)、TypeHint(接下来会讲到,用于泛型类型参数)、Java 原生 Class(例如 String.class) 等;不过字符串形式的用法即将废弃,如果确实有必要,请使用 Class.forName() 等方法来解决。
- 声明类型信息的常见手段
通过 TypeInformation.of() 方法,可以简单地创建类型信息对象。
1. 对于非泛型的类,直接传入 Class 对象即可
2. 对于泛型类,需要借助 TypeHint 来保存泛型类型信息
3. 预定义的快捷方式
例如 BasicTypeInfo,这个类定义了一系列常用类型的快捷方式,对于 String、Boolean、Byte、Short、Integer、Long、Float、Double、Char 等基本类型的类型声明,可以直接使用。
4. 自定义 TypeInfo 和 TypeInfoFactory
通过自定义 TypeInfo 为任意类提供 Flink 原生内存管理(而非 Kryo),可令存储更紧凑,运行时也更高效。
开发者在自定义类上使用 @TypeInfo 注解,随后创建相应的 TypeInfoFactory 并覆盖 createTypeInfo 方法。
注意需要继承 TypeInformation 类,为每个字段定义类型,并覆盖元数据方法,例如是否是基本类型(isBasicType)、是否是 Tuple(isTupleType)、元数(对于一维的 Row 类型,等于字段的个数)等等,从而为 TypeExtractor 提供决策依据。
更多示例,请参考 Flink 源码的 org/apache/flink/api/java/typeutils/TypeInfoFactoryTest.java
Kryo 序列化
待研究中 …