Flink 类型和序列化机制简介

妖狐艹你老母 2022-11-07 14:30 149阅读 0赞

参考文章：[Flink 类型和序列化机制简介][Flink]

使用 Flink 编写处理逻辑时，新手总是容易被林林总总的概念所混淆：

> 为什么 Flink 有那么多的类型声明方式？  
> BasicTypeInfo.STRING\_TYPE\_INFO、Types.STRING 、Types.STRING() 有何区别？  
> TypeInfoFactory 又是什么？  
> TypeInformation.of 和 TypeHint 是如何使用的呢？

接下来本文将逐步解密 Flink 的类型和序列化机制。

## Flink 的类型分类 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70][] 图 1：Flink 类型分类

Flink 的类型系统源码位于 org.apache.flink.api.common.typeinfo 包，让我们对图 1 深入追踪，看一下类的继承关系图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 1][] 图 2：TypeInformation 类继承关系图

可以看到，图 1 和 图 2 是一一对应的，TypeInformation 类是描述一切类型的公共基类，它和它的所有子类必须可序列化（Serializable），因为类型信息将会伴随 Flink 的作业提交，被传递给每个执行节点。  
由于 Flink 自己管理内存，采用了一种非常紧凑的存储格式（见[官方博文][Link 1]），因而类型信息在整个数据处理流程中属于至关重要的元数据。

## TypeExtractror 类型提取 ##

> Flink 内部实现了名为 **TypeExtractror **的类，可以利用方法签名、子类信息等蛛丝马迹，自动提取和恢复类型信息（当然也可以显式声明，即本文所介绍的内容）。
> 
> 然而由于 Java 的类型擦除，自动提取并不是总是有效。因而一些情况下（例如通过 URLClassLoader 动态加载的类），仍需手动处理；例如下图中对 DataSet 变换时，使用 .returns() 方法声明返回类型。
> 
> 这里需要说明一下，returns() 接受三种类型的参数：字符串描述的类名（例如 "String"）、TypeHint（接下来会讲到，用于泛型类型参数）、Java 原生 Class（例如 String.class) 等；不过字符串形式的用法即将废弃，如果确实有必要，请使用 Class.forName() 等方法来解决。

![20210312163212962.png][] 图 3：使用 .returns 方法声明返回类型

下面是 ExecutionEnvironment 类的 registerType 方法，它可以向 Flink 注册子类信息（Flink 认识父类，但不一定认识子类的一些独特特性，因而需要注册），下面是 Flink-ML 机器学习库代码的例子：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 2][] 图 4：Flink-ML 注册子类类型信息

从下图可以看到，如果通过 TypeExtractor.createTypeInfo(type) 方法获取到的类型信息属于 PojoTypeInfo 及其子类，那么将其注册到一起；否则统一交给 Kryo 去处理，Flink 并不过问（这种情况下性能会变差）。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 3][] 图 5：Flink 允许注册自定义类型

## 声明类型信息的常见手段 ##

通过 TypeInformation.of() 方法，可以简单地创建类型信息对象。

**1. 对于非泛型的类，直接传入 Class 对象即可**

![20210312163621604.png][] 图 6：class 对象作为参数

**2. 对于泛型类，需要借助 TypeHint 来保存泛型类型信息**

TypeHint 的原理是创建匿名子类，运行时 TypeExtractor 可以通过 getGenericSuperclass(). getActualTypeArguments() 方法获取保存的实际类型。

![2021031216365929.png][] 图 7：TypeHint 作为参数，保存泛型信息

**3. 预定义的快捷方式**

例如 BasicTypeInfo，这个类定义了一系列常用类型的快捷方式，对于 String、Boolean、Byte、Short、Integer、Long、Float、Double、Char 等基本类型的类型声明，可以直接使用。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 4][] 图 8：BasicTypeInfo 快捷方式

例如下面是对 Row 类型各字段的类型声明，使用方法非常简明，不再需要 new XxxTypeInfo<>(很多很多参数)

![20210312163808951.png][] 图 9：使用 BasicTypeInfo 快捷方式来声明一行（Row）每个字段的类型信息

当然，如果觉得 BasicTypeInfo 还是太长，Flink 还提供了完全等价的 Types 类（org.apache.flink.api.common.typeinfo.Types）：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 5][] 图 10：Types 类

特别需要注意的是，flink-table 模块也有一个 Types 类（org.apache.flink.table.api.Types），用于 table 模块内部的类型定义信息，用法稍有不同。使用 IDE 的自动 import 时一定要小心：

![20210312163858744.png][] 图 11：flink-table 模块的 Types 类

**4. 自定义 TypeInfo 和 TypeInfoFactory**

> 通过自定义 TypeInfo 为任意类提供 Flink 原生内存管理（而非 Kryo），可令存储更紧凑，运行时也更高效。
> 
> 开发者在自定义类上使用 @TypeInfo 注解，随后创建相应的 TypeInfoFactory 并覆盖 createTypeInfo 方法。
> 
> 注意需要继承 TypeInformation 类，为每个字段定义类型，并覆盖元数据方法，例如是否是基本类型（isBasicType）、是否是 Tuple（isTupleType）、元数（对于一维的 Row 类型，等于字段的个数）等等，从而为 TypeExtractor 提供决策依据。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 6][] 图 12：为自定义类提供类型支持（图片未展示全部字段）

更多示例，请参考 Flink 源码的 [org/apache/flink/api/java/typeutils/TypeInfoFactoryTest.java][org_apache_flink_api_java_typeutils_TypeInfoFactoryTest.java]

## **TypeSerializer** ##

Flink 自带了很多 TypeSerializer 子类，大多数情况下各种自定义类型都是常用类型的排列组合，因而可以直接复用：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 7][] 图 13：Flink 自带的 TypeSerializer 子类概览

如果不能满足，那么可以继承 TypeSerializer 及其子类以实现自己的序列化器。

## Kryo 序列化 ##

对于 Flink 无法序列化的类型（例如用户自定义类型，没有 registerType，也没有自定义 TypeInfo 和 TypeInfoFactory），默认会交给 Kryo 处理。

如果 Kryo 仍然无法处理（例如 Guava、Thrift、Protobuf 等第三方库的一些类），有以下两种解决方案：

1. 可以强制使用 Avro 来替代 Kryo：

env.getConfig().enableForceAvro();   // env 代表 ExecutionEnvironment 对象, 下同

2. 为 Kryo 增加自定义的 Serializer 以增强 Kryo 的功能：

env.getConfig().addDefaultKryoSerializer(Class<?> type, Class<? extends Serializer<?>> serializerClass

![20210312164053155.png][] 图 14：为 Kryo 增加自定义的 Serializer

以及

env.getConfig().registerTypeWithKryoSerializer(Class<?> type, T serializer)

![20210312164143933.png][] 图 15：为 Kryo 增加自定义的 Serializer

如果希望完全禁用 Kryo（100% 使用 Flink 的序列化机制），则可以使用以下设置，但注意一切无法处理的类都将导致异常：

env.getConfig().disableGenericTypes();

## 类型机制的陷阱与缺陷 ##

金无足赤，人无完人。Flink 内置的类型系统虽然强大而灵活，但仍然有一些需要注意的点：

**1. Lambda 函数的类型提取**

> 由于 Flink 类型提取依赖于继承等机制，而 lambda 函数比较特殊，它是匿名的，也没有与之相关的类，所以其类型信息较难获取。  
> Eclipse 的 JDT 编译器会把 lambda 函数的泛型签名等信息写入编译后的字节码中，而对于 javac 等常见的其他编译器，则不会这样做，因而 Flink 就无法获取具体类型信息了。

**2. Kryo** **的 JavaSerializer** **在 Flink 下存在 Bug**

> 推荐使用 org.apache.flink.api.java.typeutils.runtime.kryo.JavaSerializer 而非 com.esotericsoftware.kryo.serializers.JavaSerializer 以防止与 Flink 不兼容。

## 类型机制与内存管理 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 8][] 图 16：类型信息到内存块

下面以 StringSerializer 为例，来看下 Flink 是如何紧凑管理内存的：

![20210312164325442.png][] 图 17：StringSerializer 类的 serialize() 方法

下面是具体的序列化过程：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 9][] 图 18：String 对象的序列化过程

可以看到，Flink 对于内存管理是非常细致的，层次分明，代码也容易理解。

## 参考阅读 ##

[Data Types & Serialization][Data Types _ Serialization]

[Flink 原理与实现：内存管理][Flink 1]

[Flink 的数据类型和序列化][Flink 2]

[Flink]: https://my.oschina.net/u/3005325/blog/3026696
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70]: /images/20221023/318c6cc516f0454ea868ef7d77cc4089.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 1]: /images/20221023/a3cc20f825a9454a8a6af9ca098d6018.png
[Link 1]: https://flink.apache.org/news/2015/05/11/Juggling-with-Bits-and-Bytes.html
[20210312163212962.png]: /images/20221023/5bd728fdb2bc4846bd1b5784e7f27e33.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 2]: /images/20221023/7fef8dcdde1c4c5995327375757a3f31.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 3]: /images/20221023/213d2a7b582642e18f1e8ae28c0f2dfc.png
[20210312163621604.png]: /images/20221023/f4e37e529ae848d48a83e35a8d3c78e2.png
[2021031216365929.png]: /images/20221023/9f13bbe97a7144d0951884ab41aed219.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 4]: /images/20221023/dfe2b75f9c974643836be368289a4929.png
[20210312163808951.png]: /images/20221023/7ce1b2a3b2714958b294e0e0c79fc9b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 5]: /images/20221023/6c33d2a70af74b86ac7136b89323a02b.png
[20210312163858744.png]: /images/20221023/3c3b2f1d5b2747e68de9f90a5190fb94.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 6]: /images/20221023/2a0681eb3ff24bc78dba4375314833d8.png
[org_apache_flink_api_java_typeutils_TypeInfoFactoryTest.java]: https://www.oschina.net/action/GoToLink?url=https%3A%2F%2Fgithub.com%2Fapache%2Fflink%2Fblob%2Fmaster%2Fflink-core%2Fsrc%2Ftest%2Fjava%2Forg%2Fapache%2Fflink%2Fapi%2Fjava%2Ftypeutils%2FTypeInfoFactoryTest.java
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 7]: /images/20221023/e0fc68a1019043afa2ddeaba6a615eff.png
[20210312164053155.png]: /images/20221023/81eb17fcb3524f218a4692bdedf9fe6b.png
[20210312164143933.png]: /images/20221023/a87d8f0718e048b4a58721ef12f1fb49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 8]: /images/20221023/19b8dd85a0004644a3b857609526c950.png
[20210312164325442.png]: /images/20221023/d66fb611f2104762855543bd081943e8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2p1c3RscGY_size_16_color_FFFFFF_t_70 9]: /images/20221023/18d6599a58764e32b1a00e40ee87c216.png
[Data Types _ Serialization]: https://www.oschina.net/action/GoToLink?url=https%3A%2F%2Fci.apache.org%2Fprojects%2Fflink%2Fflink-docs-release-1.6%2Fdev%2Ftypes_serialization.html
[Flink 1]: https://www.oschina.net/action/GoToLink?url=http%3A%2F%2Fwuchong.me%2Fblog%2F2016%2F04%2F29%2Fflink-internals-memory-manage%2F
[Flink 2]: https://www.oschina.net/action/GoToLink?url=https%3A%2F%2Fwww.jianshu.com%2Fp%2Fe8bc484fa4c5