乐趣区

关于大数据:开源交流丨批流一体数据集成框架ChunJun数据传输模块详解分享

课件获取:关注公众号“ChunJun”,后盾私信“课件”取得直播课件

视频回放:点击这里

ChunJun 开源我的项目地址:github 丨 gitee 喜爱咱们的我的项目给咱们点个__ STAR!STAR!!STAR!!!(重要的事件说三遍)__

技术交换钉钉 qun:30537511

本期咱们带大家回顾一下六六同学的直播分享《ChunJun 数据传输模块介绍》。

一、ChunJun 数据类型转换

1、类型转换解决的问题

大家一听到「ChunJun 数据类型转换」这个概念,可能会联想到上下游之间进行数据交互时会波及到的隐式转换。如果上游和上游数据类型统一,则不须要对数据进行任何干涉,间接进行下发即可。

然而大多数状况下会波及到两个问题,一是上游的数据源类型和上游的数据源类型不统一。比方 MySql 的 varchar 类型要写到 HdfsOrc 文件里的 string 类型的话,在上游的示意是 varchar,在上游的示意是 string,但实际上两头段 java 的类型都是 string。

另外一种状况则是,上下游之间不止数据源类型不一样,数据类型也不一样,除了要做类型的映射之外,还须要对数据自身进行改变。比方,MySql 的 date 类型要写到上游 timestamp 类型,咱们须要进行的操作是把 date 中的毫秒级的工夫戳拿进去,转换成 timestamp 的类型,再往上游去写。

这样就引出了一个问题,如何建设所有数据源类型之间的映射 / 转换关系?上面将为大家解答这个问题。

2、类型映射概览

• client 端:在 Factory 类中通过 RawConverter 类建设映射关系

• source 端:将数据封装成 AbstractBaseColumn

• sink 端:通过 AbstractBaseColumn 中的转换方法将数据转换成对应类型

ChunJun 目前反对的数据类型映射关系图如下:

3、类型映射详解

以 Timestamp 为例,如果要写入到 Long 类型的话,依据上文展现的 ChunJun 数据类型映射关系图,最终映射到 TimestampColumn 中,具体流程如下图:

下面这个例子形容的是一个独自的字段,失常状况下,会解决多个字段,这时的类型映射详解状况如下图:

as 办法就是数据类型转换的办法。应用这个机制之后,在上游能够只关怀须要的数据类型,减少开发效率。

二、ChunJun 数据传输过程

理解完 ChunJun 数据类型转换后,咱们来为大家分享 ChunJun 的数据传输过程。

1、上下游数据传输方式

在 ChunJun 中进行同步作业,有两种状况,一是算子链关上的状况,上游的 Source 和上游的 Sink 会被合并成一个 task,有同一个线程去做调度;二是把算子链进行敞开,Source 和 Sink 各自造成一个 task,也有各自的线程去进行调度。

在算子链关上的状况下,上下游数据传输方式可分为两种,对象重用和拷贝。

● 对象重用

· 上下游数据传输应用办法调用的模式,将上游产生的数据的对象援用间接交给上游

· 上下游算子须要造成算子链,作业开启对象重用

· env.getConfig().enableObjectReuse();

● 拷贝

· 上游传输给上游的数据,须要通过一次深拷贝

· 上下游算子须要造成算子链

算子链的益处是能够缩小序列化的操作,那么为什么咱们还要引入序列化呢?因为 ChunJun 的特殊性。ChunJun 同步作业的话,只有上下游两个算子,且都对接了正式的数据源,读写的时候会导致线程梗塞。因而下限由网络 io 决定,如果断开算子链,cpu 会在一端线程阻塞的时候切换到另外一端。在序列化的性能较高时,线程上下文切换带来的性能降落齐全能够被补救。

通过测试,序列化的性能比对象重用和拷贝高 30% 左右。

● 序列化

· 上下游数据传输依赖于网络传输。上游数据进行序列化成 byte 数组后进行网络传输,上游收到数据后须要进行反序列化

· 上下游之间不造成算子链

晓得要做序列化后,会产生一些思考,带着这些疑难,接着往下看。

• 序列化和反序列化在什么时候产生?

• Flink 反对哪些序列化?

• 序列化是怎么做的?

• 怎么找到适宜的序列化形式?

• 如何实现自定义的序列化?

2、序列化传输过程

下图是 ChunJun 在进行序列化操作时的数据传输链路图:

3、DataOutView

4、TypeInformation 介绍

5、kryo 序列化 &BaseSerializer

同样是序列化一个 int 对象,对 kryo 来说,首先须要晓得它的类型,而后从高位到低位顺次去写入。

DataOutputView 则是间接调用一个 writeInt 的办法,写一句要害代码即可:

UNSAFE.putInt(

this.buffer,

BASE_OFFSET + this.position, v);

三、ChunJun 序列化实现

1、ColumnRowData 序列化过程

ColumnRowData 序列化过程采取标记位 + 理论数据的形式,具体流程如下图:

绝对于 kryo 的序列化来说:

· 实现了更密集的存储

· 兼容 null 值

· 缩小了不必要的数据传输

2、BinaryRowData 构造

因为数据区一格只占 8 个字节,且每个 index 只能占到一位,所以必定存在一些没法存储在 8 字节范畴之内的数据,可变长度局部就是用来存放数据区无奈寄存的数据。

3、BinaryRowData-setNull 操作

看到上文的 null 值判断区,有些同学可能会好奇这是什么,又是怎么进行操作的。下图将对一个下标为 11 的数据去做 setnull 操作,进行简略介绍:

4、BinaryRowData 数据存储形式

袋鼠云开源框架钉钉技术交换 qun(30537511),欢送对大数据开源我的项目有趣味的同学退出交换最新技术信息,开源我的项目库地址:https://github.com/DTStack

退出移动版