当前位置: 首页 > news >正文

Flink 数据序列化

为 Flink 量身定制的序列化框架

大家都知道现在大数据生态非常火,大多数技术组件都是运行在JVM上的,Flink也是运行在JVM上,基于JVM的数据分析引擎都需要将大量的数据存储在内存中,这就不得不面临JVM的一些问题,比如Java对象存储密度较低等。针对这些问题,最常用的方法就是实现一个显式的内存管理,也就是说用自定义的内存池来进行内存的分配回收,接着将序列化后的对象存储到内存块中。
现在Java生态圈中已经有许多序列化框架,比如说Java serialization, Kryo,Apache Avro等等。但是Flink依然是选择了自己定制的序列化框架,那么到底有什么意义呢?若Flink选择自己定制的序列化框架,对类型信息了解越多,可以在早期完成类型检查,更好的选取序列化方式,进行数据布局,节省数据的存储空间,直接操作二进制数据。
在这里插入图片描述Flink在其内部构建了一套自己的类型系统,Flink现阶段支持的类型分类如图所示,从图中可以看到Flink类型可以分为基础类型Basic、数组Arrays、复合类型Composite、辅助类型Auxiliary、泛型和其它类型GenericFlink支持任意的Java或是Scala类型。不需要像Hadoop一样去实现一个特定的接口org.apache.hadoop.io.WritableFlink能够自动识别数据类型。
在这里插入图片描述
TypeInformation的思维导图如图所示,从图中可以看出,在Flink中每一个具体的类型都对应了一个具体的TypeInformation实现类,例如BasicTypeInformation中的IntegerTypeInformationFractionalTypeInformation都具体的对应了一个TypeInformation。然后还有BasicArrayTypeInformationCompositeType以及一些其它类型,也都具体对应了一个TypeInformation

TypeInformationFlink类型系统的核心类。对于用户自定义的Function来说,Flink需要一个类型信息来作为该函数的输入输出类型,即TypeInfomation。该类型信息类作为一个工具来生成对应类型的序列化器TypeSerializer,并用于执行语义检查,比如当一些字段在作为joingrouping的键时,检查这些字段是否在该类型中存在。

Flink 的序列化过程

Flink序列化过程中,进行序列化操作必须要有序列化器,那么序列化器从何而来?
每一个具体的数据类型都对应一个TypeInformation的具体实现,每一个TypeInformation都会为对应的具体数据类型提供一个专属的序列化器。通过 Flink的序列化过程图可以看到TypeInformation会提供一个createSerialize()方法,通过这个方法就可以得到该类型进行数据序列化操作与反序化操作的对象TypeSerializer
在这里插入图片描述对于大多数数据类型 Flink可以自动生成对应的序列化器,能非常高效地对数据集进行序列化和反序列化,比如,BasicTypeInfoWritableTypeInfo等,但针对GenericTypeInfo类型,Flink会使用Kyro进行序列化和反序列化。其中,TuplePojoCaseClass类型是复合类型,它们可能嵌套一个或者多个数据类型。在这种情况下,它们的序列化器同样是复合的。它们会将内嵌类型的序列化委托给对应类型的序列化器。

简单的介绍下Pojo的类型规则,即在满足一些条件的情况下,才会选用Pojo的序列化进行相应的序列化与反序列化的一个操作。即类必须是Public的,且类有一个public的无参数构造函数,该类(以及所有超类)中的所有非静态no-static、非瞬态no-transient字段都是public的(和非最终的final)或者具有公共gettersetter方法,该方法遵循gettersetterJava bean命名约定。当用户定义的数据类型无法识别为POJO类型时,必须将其作为GenericType处理并使用Kryo进行序列化。

Flink自带了很多TypeSerializer子类,大多数情况下各种自定义类型都是常用类型的排列组合,因而可以直接复用,如果内建的数据类型和序列化方式不能满足你的需求,Flink的类型信息系统也支持用户拓展。若用户有一些特殊的需求,只需要实现 TypeInformationTypeSerializerTypeComparator即可定制自己类型的序列化和比较大小方式,来提升数据类型在序列化和比较时的性能。

序列化就是将数据结构或者对象转换成一个二进制串的过程,在Java里面可以简单地理解成一个byte数组。而反序列化恰恰相反,就是将序列化过程中所生成的二进制串转换成数据结构或者对象的过程。下面就以内嵌型的Tuple3这个对象为例,简述一下它的序列化过程。
[点击并拖拽以移动] ​

Tuple3包含三个层面,一是int类型,一是double类型,还有一个是PersonPerson包含两个字段,一是int型的ID,另一个是 String 类型的name,它在序列化操作时,会委托相应具体序列化的序列化器进行相应的序列化操作。从图中可以看到Tuple3 会把 int类型通过IntSerializer进行序列化操作,此时int只需要占用四个字节就可以了。根据int占用四个字节,这个能够体现出Flink可序列化过程中的一个优势,即在知道数据类型的前提下,可以更好的进行相应的序列化与反序列化操作。相反,如果采用Java的序列化,虽然能够存储更多的属性信息,但一次占据的存储空间会受到一定的损耗。Person类会被当成一个Pojo对象来进行处理,PojoSerializer序列化器会把一些属性信息使用一个字节存储起来。同样,其字段则采取相对应的序列化器进行相应序列化,在序列化完的结果中,可以看到所有的数据都是由MemorySegment去支持。

MemorySegment具有什么作用呢? MemorySegmentFlink中会将对象序列化到预分配的内存块上,它代表1个固定长度的内存,默认大小为32 kbMemorySegment代表Flink中的一个最小的内存分配单元,相当于是Java的一个byte数组。 每条记录都会以序列化的形式存储在一个或多个MemorySegment中。

Flink 序列化的最佳实践

Flink常见的应用场景有四种,即注册子类型、注册自定义序列化器、添加类型提示、手动创建TypeInformation,具体如下:
【1】注册子类型: 如果函数签名只描述了超类型,但是它们实际上在执行期间使用了超类型的子类型,那么让Flink了解这些子类型会大大提高性能。可以在StreamExecutionEnvironmentExecutionEnvironment中调用.registertype (clazz) 注册子类型信息。
【2】注册自定义序列化器: 对于不适用于自己的序列化框架的数据类型,Flink会使用Kryo来进行序列化,并不是所有的类型都与Kryo无缝连接,具体注册方法在下文介绍。
【3】添加类型提示: 有时,当Flink用尽各种手段都无法推测出泛型信息时,用户需要传入一个类型提示TypeHint,这个通常只在Java API中需要。
【4】手动创建TypeInformation 在某些API调用中,这可能是必需的,因为Java的泛型类型擦除导致Flink无法推断数据类型。
其实在大多数情况下,用户不必担心序列化框架和注册类型,因为Flink已经提供了大量的序列化操作,不需要去定义自己的一些序列化器,但是在一些特殊场景下,需要去做一些相应的处理。

实践 - 类型声明: 类型声明去创建一个类型信息的对象是通过哪种方式?通常是用TypeInformation.of()方法来创建一个类型信息的对象,具体说明如下:
【1】对于非泛型类,直接传入class对象即可

PojoTypeInfo<Person> typeInfo = (PojoTypeInfo<Person>) TypeInformation.of(Person.class);

【2】对于泛型类,需要通过TypeHint来保存泛型类型信息

final TypeInfomation<Tuple2<Integer,Integer>> resultType = TypeInformation.of(new TypeHint<Tuple2<Integer,Integer>>(){});

【3】预定义常量:BasicTypeInfo,这个类定义了一系列常用类型的快捷方式,对于StringBooleanByteShortIntegerLongFloatDoubleChar等基本类型的类型声明,可以直接使用。而且Flink还提供了完全等价的Typesorg.apache.flink.api.common.typeinfo.Types。特别需要注意的是,flink-table模块也有一个Typesorg.apache.flink.table.api.Types,用于table模块内部的类型定义信息,用法稍有不同。使用IDE 的自动import时一定要小心。
【4】自定义TypeInfoTypeInfoFactory 通过自定义TypeInfo为任意类提供Flink原生内存管理(而非Kryo),使存
储更紧凑,运行时也更高效。需要注意在自定义类上使用@TypeInfo注解,随后创建相应的TypeInfoFactory并覆盖createTypeInfo()方法。

@TypeInfo(MyTupleTypeInfoFactory.class)
public class MyTuple<T0,T1>{public T0 myfield0;public T1 myfield1;
}public class MyTupleTypeInfoFactory extends TypeInfoFactory<MyTuple>{@Overridepublic TypeInformation<MyTuple> createTypeInfo(Type t, Map<String, TypeInfomation<?>> genericParameters){return new MyTupleTypeInfo(genericParameters.get("T0").genericParameters.get("T1"));}
}

实践 - 注册子类型

Flink认识父类,但不一定认识子类的一些独特特性,因此需要单独注册子类型。StreamExecutionEnvironmentExecutionEnvironment提供registerType()方法用来向Flink注册子类信息。

final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.registerType(typeClass);

registerType()方法内部,会使用TypeExtractor来提取类型信息,如上所示,获取到的类型信息属于PojoTypeInfo及其子类,那么需要将其注册到一起,否则统一交给Kryo去处理,Flink并不过问 ( 这种情况下性能会变差 )。

实践 - Kryo 序列化

对于Flink无法序列化的类型(例如用户自定义类型,没有registerType,也没有自定义TypeInfoTypeInfoFactory),默认会交给 Kryo处理,如果Kryo仍然无法处理(例如GuavaThriftProtobuf等第三方库的一些类),有两种解决方案:
【1】强制使用Avro来代替Kryo

env.getConfig().enableForceAvro();

【2】为Kryo增加自定义的Serializer以增强Kryo的功能

env.getConfig().addDefaultKryoSerializer(clazz, serializer);

注:如果希望完全禁用Kryo100%使用Flink的序列化机制),可以通过Kryoenv.getConfig().disableGenericTypes()的方式完成,但注意一切无法处理的类都将导致异常,这种对于调试非常有效。

Flink 通信层的序列化

FlinkTask之间如果需要跨网络传输数据记录, 那么就需要将数据序列化之后写入NetworkBufferPool,然后下层的Task读出之后再进行反序列化操作,最后进行逻辑处理。为了使得记录以及事件能够被写入 Buffer,随后在消费时再从Buffer中读出,
Flink提供了数据记录序列化器RecordSerializer与反序列化器RecordDeserializer以及事件序列化器EventSerializer

Function发送的数据被封装成SerializationDelegate,它将任意元素公开为IOReadableWritable以进行序列化,通过setInstance()来传入要序列化的数据。在Flink通信层的序列化中,有几个问题值得关注,具体如下:
【1】何时确定Function的输入输出类型?
[点击并拖拽以移动] ​

在构建StreamTransformation的时候通过TypeExtractor工具确定Function的输入输出类型。TypeExtractor类可以根据方法签名、子类信息等蛛丝马迹自动提取或恢复类型信息。
【2】何时确定Function的序列化 / 反序列化器?
构造StreamGraph时, 通过TypeInfomationcreateSerializer()方法获取对应类型的序列化器TypeSerializer,并在addOperator()的过程中执行setSerializers() 操作,设置StreamConfigTYPESERIALIZERIN1TYPESERIALIZERIN2TYPESERIALIZEROUT_1属性。
【3】何时进行真正的序列化 / 反序列化操作? 这个过程与TypeSerializer又是怎么联系在一起的呢?
构造StreamGraph时, 通过TypeInfomationcreateSerializer()方法获取对应类型的序列化器TypeSerializer,并在addOperator()的过程中执行setSerializers()操作,设置StreamConfigTYPESERIALIZERIN1TYPESERIALIZERIN2TYPESERIALIZEROUT_1属性。
【4】何时进行真正的序列化 / 反序列化操作? 这个过程与TypeSerializer又是怎么联系在一起的呢?
[点击并拖拽以移动] ​

大家都应该清楚TaskStreamTask两个概念,Task是直接受TaskManager管理和调度的,而Task又会调用StreamTask,而StreamTask中真正封装了算子的处理逻辑。在run()方法中,首先将反序列化后的数据封装成StreamRecord交给算子处理;然后将处理结果通过Collector发送给下游 ( 在构建Collector时已经确定了SerializtionDelegate),并通过RecordWriter写入器将序列化后的结果写入DataOutput;最后序列化的操作交给SerializerDelegate处理,实际还是通过TypeSerializerserialize()方法完成。

相关文章:

Flink 数据序列化

为 Flink 量身定制的序列化框架 大家都知道现在大数据生态非常火&#xff0c;大多数技术组件都是运行在JVM上的&#xff0c;Flink也是运行在JVM上&#xff0c;基于JVM的数据分析引擎都需要将大量的数据存储在内存中&#xff0c;这就不得不面临JVM的一些问题&#xff0c;比如Ja…...

【并发设计模式】聊聊两阶段终止模式如何优雅终止线程

在软件设计中&#xff0c;抽象出了23种设计模式&#xff0c;用以解决对象的创建、组合、使用三种场景。在并发编程中&#xff0c;针对线程的操作&#xff0c;也抽象出对应的并发设计模式。 两阶段终止模式- 优雅停止线程避免共享的设计模式- 只读、Copy-on-write、Thread-Spec…...

Java实现非对称加密【详解】

Java实现非对称加密 1. 简介2. 非对称加密算法--DH&#xff08;密钥交换&#xff09;3. 非对称加密算法--RSA非对称加密算法--EIGamal5. 总结6 案例6.1 案例16.2 案例2 1. 简介 公开密钥密码学&#xff08;英语&#xff1a;Public-key cryptography&#xff09;也称非对称式密…...

simulinkveristandlabview联合仿真——模型导入搭建人机界面

目录 1.软件版本 2.搭建simulink仿真模型 编译错误 3.导入veristand并建立工程 4.veristand导入labview labview显示veristand工程数据 labview设置veristand工程数据 运行labview工程 1.软件版本 matlab2020a&#xff0c;veristand2020 R4&#xff0c;labview2020 SP…...

k8s中Helm工具实践

k8s中Helm工具实践 1&#xff09;安装redis-cluster 先搭建一个NFS的SC&#xff08;只需要SC&#xff0c;不需要pvc&#xff09;&#xff0c;具体步骤此文档不再提供&#xff0c;请参考前面相关章节。 下载redis-cluster的chart包 helm pull bitnami/redis-cluster --untar…...

推荐算法架构7:特征工程(吊打面试官,史上最全!)

系列文章&#xff0c;请多关注 推荐算法架构1&#xff1a;召回 推荐算法架构2&#xff1a;粗排 推荐算法架构3&#xff1a;精排 推荐算法架构4&#xff1a;重排 推荐算法架构5&#xff1a;全链路专项优化 推荐算法架构6&#xff1a;数据样本 推荐算法架构7&#xff1a;特…...

Web前端 ---- 【Vue】vue路由守卫(全局前置路由守卫、全局后置路由守卫、局部路由path守卫、局部路由component守卫)

目录 前言 全局前置路由守卫 全局后置路由守卫 局部路由守卫之path守卫 局部路由守卫之component守卫 前言 本文介绍Vue2最后的知识点&#xff0c;关于vue的路由守卫。也就是鉴权&#xff0c;不是所有的组件任何人都可以访问到的&#xff0c;需要权限&#xff0c;而根据权限…...

uniapp点击tabbar之前做判断

在UniApp中&#xff0c;可以通过监听 tabBar 的 click 事件来在点击 tabBar 前做判断。具体步骤如下&#xff1a; 在 pages.json 文件中配置 tabBar&#xff0c;例如&#xff1a; {"pages":[{"path":"pages/home/home","name":"h…...

DLLNotFoundException:xxx tolua... 错误打印

DLLNotFoundException:xxx tolua... 错误打印 一、DLLNotFoundException介绍二、Plugins文件夹文件目录结构如下&#xff1a; 三、Plugins中的Android文件夹四、Plugins中的IOS文件夹这里不说了没测试过不过原理应该也是选择对应的平台即可五、Plugins中的x86和X86_64文件夹 一…...

Python量化投资——金融数据最佳实践: 使用qteasy+tushare搭建本地金融数据仓库并定期批量更新【附源码】

用qteasytushare实现金融数据本地化存储及访问 目的什么是qteasy什么是tushare为什么要本地化使用qteasy创建本地数据仓库qteasy支持的几种本地化仓库类型配置本地数据仓库配置tushare 的API token 配置本地数据源 —— 用MySQL数据库作为本地数据源下载金融历史数据 数据的定期…...

【投稿】北海 - Rust与面向对象(二)

模板方法 Rust提供了trait&#xff0c;类似于面向对象的接口&#xff0c;不同的是&#xff0c;将传统面向对象的虚函数表从对象中分离出来&#xff0c;trait仍然是一个函数表&#xff0c;只不过是独立的&#xff0c;它的参数self指针可以指向任何实现了该trait的结构。 从对象中…...

HarmonyOS构建第一个ArkTS应用(FA模型)

构建第一个ArkTS应用&#xff08;FA模型&#xff09; 创建ArkTS工程 若首次打开DevEco Studio&#xff0c;请点击Create Project创建工程。如果已经打开了一个工程&#xff0c;请在菜单栏选择File > New > Create Project来创建一个新工程。 选择Application应用开发&a…...

阿里云 ARMS 应用监控重磅支持 Java 21

作者&#xff1a;牧思 & 山猎 前言 今年的 9 月 19 日&#xff0c;作为最新的 LTS (Long Term Support) Java 版本&#xff0c;Java 21 正式 GA&#xff0c;带来了不少重量级的更新&#xff0c;详情请参考 The Arrival of Java 21 [ 1] 。虽然目前 Java 11 和 Java 17 都…...

C++ 类的析构函数和构造函数

构造函数 类的构造函数是类的一种特殊的成员函数&#xff0c;它会在每次创建类的新对象时执行。主要用来在创建对象时初始化对象即为对象成员变量赋初始值。 构造函数的名称与类的名称是完全相同的&#xff0c;并且不会返回任何类型&#xff0c;也不会返回 void。构造函数可用…...

STM32——CAN协议

文章目录 一.CAN协议的基本特点1.1 特点1.2 电平标准1.3 基本的五个帧1.4 数据帧 二.数据帧解析2.1 帧起始和仲裁段2.2 控制段2.3 数据段和CRC段2.4 ACK段和帧结束 三.总线仲裁四.位时序五.STM32CAN控制器原理与配置5.1 STM32CAN控制器介绍5.2 CAN的模式5.3 CAN框图 六 手册寄存…...

数据结构-如何巧妙实现一个栈?逐步解析与代码示例

文章目录 引言1.栈的基本概念2.选择数组还是链表&#xff1f;3. 定义栈结构4.初始化栈5.压栈操作6.弹栈操作7.查看栈顶和判断栈空9.销毁栈操作10.测试并且打印栈内容栈的实际应用结论 引言 栈是一种基本但强大的数据结构&#xff0c;它在许多算法和系统功能中扮演着关键角色。…...

web前端之拖拽API、vue3实现图片上传拖拽排序、拖放、投掷、复制、若依、vuedraggable

MENU vue2html5原生dom原生JavaScript实现跨区域拖放vue2实现跨区域拖放vue2mousedown实现全屏拖动&#xff0c;全屏投掷vue3element-plusvuedraggable实现图片上传拖拽排序vue2transition-group实现拖动排序原生拖拽排序 vue2html5原生dom原生JavaScript实现跨区域拖放 关键代…...

第11章 GUI Page403~405 步骤三 设置滚动范围

运行效果&#xff1a; 源代码&#xff1a; /**************************************************************** Name: wxMyPainterApp.h* Purpose: Defines Application Class* Author: yanzhenxi (3065598272qq.com)* Created: 2023-12-21* Copyright: yanzhen…...

【Spring Security】打造安全无忧的Web应用--使用篇

&#x1f973;&#x1f973;Welcome Huihuis Code World ! !&#x1f973;&#x1f973; 接下来看看由辉辉所写的关于Spring Security的相关操作吧 目录 &#x1f973;&#x1f973;Welcome Huihuis Code World ! !&#x1f973;&#x1f973; 一.Spring Security中的授权是…...

体验一下 CodeGPT 插件

体验一下 CodeGPT 插件 0. 背景1. CodeGPT 插件安装2. CodeGPT 插件基本配置3. (可选)CodeGPT 插件预制提示词原始配置(英文)4. CodeGPT 插件预制提示词配置(中文)5. 简单验证一下 0. 背景 看到B站Up主 “wwwzhouhui” 一个关于 CodeGPT 的视频&#xff0c;感觉挺有意思&#…...

网络编程(Modbus进阶)

思维导图 Modbus RTU&#xff08;先学一点理论&#xff09; 概念 Modbus RTU 是工业自动化领域 最广泛应用的串行通信协议&#xff0c;由 Modicon 公司&#xff08;现施耐德电气&#xff09;于 1979 年推出。它以 高效率、强健性、易实现的特点成为工业控制系统的通信标准。 包…...

聊聊 Pulsar:Producer 源码解析

一、前言 Apache Pulsar 是一个企业级的开源分布式消息传递平台&#xff0c;以其高性能、可扩展性和存储计算分离架构在消息队列和流处理领域独树一帜。在 Pulsar 的核心架构中&#xff0c;Producer&#xff08;生产者&#xff09; 是连接客户端应用与消息队列的第一步。生产者…...

自然语言处理——循环神经网络

自然语言处理——循环神经网络 循环神经网络应用到基于机器学习的自然语言处理任务序列到类别同步的序列到序列模式异步的序列到序列模式 参数学习和长程依赖问题基于门控的循环神经网络门控循环单元&#xff08;GRU&#xff09;长短期记忆神经网络&#xff08;LSTM&#xff09…...

html-<abbr> 缩写或首字母缩略词

定义与作用 <abbr> 标签用于表示缩写或首字母缩略词&#xff0c;它可以帮助用户更好地理解缩写的含义&#xff0c;尤其是对于那些不熟悉该缩写的用户。 title 属性的内容提供了缩写的详细说明。当用户将鼠标悬停在缩写上时&#xff0c;会显示一个提示框。 示例&#x…...

C++使用 new 来创建动态数组

问题&#xff1a; 不能使用变量定义数组大小 原因&#xff1a; 这是因为数组在内存中是连续存储的&#xff0c;编译器需要在编译阶段就确定数组的大小&#xff0c;以便正确地分配内存空间。如果允许使用变量来定义数组的大小&#xff0c;那么编译器就无法在编译时确定数组的大…...

IP如何挑?2025年海外专线IP如何购买?

你花了时间和预算买了IP&#xff0c;结果IP质量不佳&#xff0c;项目效率低下不说&#xff0c;还可能带来莫名的网络问题&#xff0c;是不是太闹心了&#xff1f;尤其是在面对海外专线IP时&#xff0c;到底怎么才能买到适合自己的呢&#xff1f;所以&#xff0c;挑IP绝对是个技…...

springboot整合VUE之在线教育管理系统简介

可以学习到的技能 学会常用技术栈的使用 独立开发项目 学会前端的开发流程 学会后端的开发流程 学会数据库的设计 学会前后端接口调用方式 学会多模块之间的关联 学会数据的处理 适用人群 在校学生&#xff0c;小白用户&#xff0c;想学习知识的 有点基础&#xff0c;想要通过项…...

MySQL JOIN 表过多的优化思路

当 MySQL 查询涉及大量表 JOIN 时&#xff0c;性能会显著下降。以下是优化思路和简易实现方法&#xff1a; 一、核心优化思路 减少 JOIN 数量 数据冗余&#xff1a;添加必要的冗余字段&#xff08;如订单表直接存储用户名&#xff09;合并表&#xff1a;将频繁关联的小表合并成…...

快刀集(1): 一刀斩断视频片头广告

一刀流&#xff1a;用一个简单脚本&#xff0c;秒杀视频片头广告&#xff0c;还你清爽观影体验。 1. 引子 作为一个爱生活、爱学习、爱收藏高清资源的老码农&#xff0c;平时写代码之余看看电影、补补片&#xff0c;是再正常不过的事。 电影嘛&#xff0c;要沉浸&#xff0c;…...

OD 算法题 B卷【正整数到Excel编号之间的转换】

文章目录 正整数到Excel编号之间的转换 正整数到Excel编号之间的转换 excel的列编号是这样的&#xff1a;a b c … z aa ab ac… az ba bb bc…yz za zb zc …zz aaa aab aac…; 分别代表以下的编号1 2 3 … 26 27 28 29… 52 53 54 55… 676 677 678 679 … 702 703 704 705;…...