intellij idea安装 js快速排序 springcloud 分布式服务 function recursion git视频教程 mac虚拟打印机 java两个数组合并 pr序列设置哪个好 mysql统计数量 spark大数据处理技术 centos查看python版本 二分查找python java时间戳转换成时间 java安装环境 java获取月份 java初级入门教程 java学习文档 java成员变量 java生成文件 java获取数据类型 javasocket java语言是什么 java怎么输出数组 java接口调用 javascript基础 只狼鬼佛 js图片上传 狮子狗皮肤 t470拆机 c4d挤压怎么用 cdr怎么画波浪线 js获取子元素 cad乘号 梦想世界科举答案 bin文件编辑器 g4560配什么显卡 熊猫头表情包制作 服务器备份软件
当前位置: 首页 > 学习教程  > 编程语言

Flink 类型和序列化机制

2020/8/31 12:17:02 文章标签:

 序列化概念

当两个进程在进行远程通信时,彼此可以发送各种类型的数据。无论是何种类型的数据,都会以二进制序列的形式在网络上传送。发送方需要把这个Java对象转换为字节序列,才能在网络上传送;接收方则需要把字节序列再恢复为Java对象。
把Java对象转换为字节序列的过程称为对象的序列化。
把字节序列恢复为Java对象的过程称为对象的反序列化。
对象的序列化主要有两种用途:
1) 把对象的字节序列永久地保存到硬盘上,通常存放在一个文件中;
2) 在网络上传送对象的字节序列。
序列化的好处:减少数据在内存、硬盘中的占用空间,减少网络传输开销,精确推算内存使用情况,降低GC频率

大多数情况下,用户不用担心flink的序列化框架,flink可以自己推断出数据的类型信息,不能推断的则采用kryo或者其他方式序列化
类型推断->自带的类型系统来处理->kryo->其他

Flink支持非常完善的数据类型,数据类型的描述信息都是由TypeInformation定义。TypeInformation主要作用是在分布式计算过程中对数据的数据类型进行管理和推断

多数情况下用户无需关系类型处理,用户与flink的数据类型处理包括:
注册子类型、注册自定义序列化、添加类型提示、手动创建typeinformation

Flink 的类型分类

 

图 1:Flink 类型分类

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

 

图 2:TypeInformation 类继承关系图

可以看到,图 1 和 图 2 是一一对应的,TypeInformation 类是描述一切类型的公共基类,它和它的所有子类必须可序列化(Serializable),因为类型信息将会伴随 Flink 的作业提交,被传递给每个执行节点。

由于 Flink 自己管理内存,采用了一种非常紧凑的存储格式(见官方博文),因而类型信息在整个数据处理流程中属于至关重要的元数据。

TypeExtractror 类型提取

Flink 内部实现了名为 TypeExtractror 的类,可以利用方法签名、子类信息等蛛丝马迹,自动提取和恢复类型信息(当然也可以显式声明,即本文所介绍的内容)。

然而由于 Java 的类型擦除,自动提取并不是总是有效。因而一些情况下(例如通过 URLClassLoader 动态加载的类),仍需手动处理;例如下图中对 DataSet 变换时,使用 .returns() 方法声明返回类型。

这里需要说明一下,returns() 接受三种类型的参数:字符串描述的类名(例如 "String")、TypeHint(接下来会讲到,用于泛型类型参数)、Java 原生 Class(例如 String.class) 等;不过字符串形式的用法即将废弃,如果确实有必要,请使用 Class.forName() 等方法来解决。

 

图 3:使用 .returns 方法声明返回类型

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

 

图 4:Flink-ML 注册子类类型信息

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

图 5:Flink 允许注册自定义类型

声明类型信息的常见手段

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

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

图 6:class 对象作为参数

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

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

图 7:TypeHint 作为参数,保存泛型信息

3. 预定义的快捷方式

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

图 8:BasicTypeInfo 快捷方式

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

图 9:使用 BasicTypeInfo 快捷方式来声明一行(Row)每个字段的类型信息

当然,如果觉得 BasicTypeInfo 还是太长,Flink 还提供了完全等价的 Types 类(org.apache.flink.api.common.typeinfo.Types):

图 10:Types 类

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

图 11:flink-table 模块的 Types 类

4. 自定义 TypeInfo 和 TypeInfoFactory

通过自定义 TypeInfo 为任意类提供 Flink 原生内存管理(而非 Kryo),可令存储更紧凑,运行时也更高效。

开发者在自定义类上使用 @TypeInfo 注解,随后创建相应的 TypeInfoFactory 并覆盖 createTypeInfo 方法。

注意需要继承 TypeInformation 类,为每个字段定义类型,并覆盖元数据方法,例如是否是基本类型(isBasicType)、是否是 Tuple(isTupleType)、元数(对于一维的 Row 类型,等于字段的个数)等等,从而为 TypeExtractor 提供决策依据。

图 12:为自定义类提供类型支持(图片未展示全部字段)

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

TypeSerializer

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

图 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

图 14:为 Kryo 增加自定义的 Serializer

以及

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

图 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 不兼容。

类型机制与内存管理

图 16:类型信息到内存块

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

图 17:StringSerializer 类的 serialize() 方法

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

图 18:String 对象的序列化过程

本文转载自:https://cloud.tencent.com/developer/article/1240444?fromSource=waitui


本文链接: http://www.dtmao.cc/news_show_150012.shtml

附件下载

相关教程

    暂无相关的数据...

共有条评论 网友评论

验证码: 看不清楚?