专栏首页工作笔记精华Flink 类型和序列化机制简介 转

Flink 类型和序列化机制简介 转

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

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

接下来本文将逐步解密 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 对象的序列化过程

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

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Java访问带有Kerberos认证的HBase

    stys35
  • 你担心spring容器中scope为prototype的bean太大内存溢出吗?

    出假设 之前一直担心spring的scope为prototype的bean在一些高并发的场景下,吃不消吗,甚至会内存溢出,这样的担心不是没有道理的,(以下是假...

    stys35
  • Vue Cli 3 搭建一个可按需引入组件的组件库架子

    babel-plugin-component 就是 Element UI 用来实现组件按需加载的一个 babel 插件。我们把它用到我们的组件库上,就不需要重新...

    stys35
  • Flink 全链路端到端延迟的测量方法

    FLink Job端到端延迟是一个重要的指标,用来衡量Flink任务的整体性能和响应延迟(大部分流式应用,要求低延迟特性)。

    zhisheng
  • 4个步骤让Flink应用程序达到生产状态

    这篇文章阐述了 Flink 应用程序达到生产状态所必须的配置步骤。在以下部分中,我们概述了在 Flink 作业达到生产状态之前技术领导、DevOps、工程师们需...

    smartsi
  • 三歪鸽了一个季度的Flink入门教程

    最近公司要把Storm集群给下线啦,所以我们都得把Storm的任务都改成Flink。

    Java3y
  • 为什么要学 Flink,Flink 香在哪?

    知道大数据的同学也应该知道 Flink 吧,最近在中国的热度比较高,在社区的推动下,Flink 技术栈在越来越多的公司开始得到应用。

    数据社
  • 9 个可以快速掌握的 Java 性能调优技巧

    大多数开发者认为性能优化是一个复杂的话题,它需要大量的工作经验和相关知识理论。好吧,这也不完全错。

    zhisheng
  • B站实时平台的架构与实践

    本文来自B站实时平台负责人郑志升在 Flink Forward Asia 2019 上的技术分享,重点介绍了B站基于 Apache Flink 的流式计算平台建...

    大数据技术架构
  • Flink Forward Asia 2019 会议所有 PPT 下载

    11 月 28 - 30 日,北京迎来了入冬以来的第一场雪,2019 Flink Forward Asia(FFA)也在初雪的召唤下顺利拉开帷幕。尽管天气寒冷,...

    zhisheng

扫码关注云+社区

领取腾讯云代金券