00:00
好,那接下来我们聊第三块内容,那么第三块呢是这样子的,大家还记得之前我们data stream API的方式实现的一个打印结果,我们用的是它自带的这个。反吸收化的方式对吧,它打印结果呢,是这个样子,也就是说其实这个东西啊,来看一下它是用的两个。对象的to string方法做了一个拼接,但是这种格式呢,对于我们下游处理很不方便,所以呢,我们需要自定义一个反修化器,对这个数据呢做一个加工,对吧?好,那这个就比较简单一点来,我们把这个分这个CDC。我再复制一份啊,就是之前呢,我们就不动了,叫二。角零了可以吧,呃,诶我应该是复制过来,我怎么是这样子,好CTRLC啊,然后CTRLV。保存一下好,呃,那接下来呢,我们在这个地方再写一个类叫方式。Yeah。
01:02
我们要制定一个什么东西呢,叫这个,所以我就不写这个什么division了,对吧,我们制定一个第civilization的一个。反消化器对吧。STEM啊,按照他的一个名字去写就好了。呃,那这样我们要制定一个。哎,这个地方。稍等一下,那我们把这个E,刚刚我少了一个E对吧,那这个改一下啊,改一下之后呢,我们把这个文件名改一下就好了,刚才少敲了一个E啊呃,那这个时候呢,我们要自定义的话,肯定要实现一个接口,或者说是。抽象类对吧,记成一个抽象类或者实现一个接口,好,那这个时候呢,我们就看它这个怎么写的啊,模仿一下它就好了,对不对啊,那它呢,实现的这个接口OK,那我们就把它拿过来啊,有时候我们写东西的时候呢,就比较简单一点啊啊把它拿过来啊在回车实现里边的一个方法,两个方法一个是获取类型,一个呢是反修化的方法,里边呢是s code,就是一行数据,然后呢有一个connect啊呃,Li代码比较熟悉的同学呢,应该知道connect将。
02:13
帮助我们可以数据输出的对吧?啊,那既然我们都是string类型,所以呢,完全我可以拿它这个代码拿过来。直接照抄对吧,因为我也是string类型,我也要返回string啊,那接下来我们要思考我们最核心的方法呢,是这个第ization啊,那我们要想一个问题,就是当前我们需要把这个数据变成什么样子。啊,变成一个什么样子,对吧,这个呢,我就不了。呃,那我们希望呢,变成一个大的一个阶层对象。杰森,里边呢,我们想一下在里边我们要什么东西?首先DB是不是得要我们那个库名得保留,因为未来我们可能在data stream当中,它可以监控多个库,所以库名要区分,有区分作用,对吧?那既然库名要,那表明要不要,那肯定要的,对不对?好,Table name我们要保留一下。
03:09
这两个,然后接下来呢,我们数据啊,数据本身也要,但是在我们之前测试环境当中,我们看出来它数据呢,分为。新增数据只有after。Delete的数据只有before,而更新数据既有after又有before,那很简单,我们就把这个数据怎么样都保留下来,哎,Before数据好,然后同时呢,我保留一个阿的数据。这个呢,我也保留下来啊,那还有在我们数据呢,其实到这块就够了,因为我们可以判断before跟after是否为来决定这个数据到底它是什么类型啊,但是呢,我们就希望把这个操作类型呢,我也给它放进去,也就是说c rud到底是什么东西,我们也放进来。对吧,那其实基本的东西呢,我们就够了啊,那基本上呢,我们就要放这些信息啊,比较简单一点,但是注意啊,在我们一行数据里边,它应该是一个比较复杂的内容,所以呢,此时这边它应该也是一个。
04:14
阶算格式,大家想一下是不是啊,也就是说在我们单场数据当中,它还是一个。结算数据嘛,对吧?啊,这里边呢,比方说有ID,它是1001,还有name啊,其他的点点点啊,类似于这样子的啊,那下面这个一样,因为数据格式是一样的啊,那我们把它拿进来啊,最终呢,我们要把它变成这个样子,因为变成这个形式之后,下游就非常方便处理了。就非常方便处理了,行,那我们接下来写这块的一个代码啊,核心的,呃,那我们要将数据呢,变成一个接送格式,首先我们要在这边创建杰森对象,用于封装结果数据,对吧?好,又有一个杰森object。
05:04
那我们要发杰森包下的得到一个result,最后呢,我们先把输出写好啊,就是大的架子搭起来输出数据,输出数据呢,我们就connector.connect要一个string,那就result.to Jason string啊Jason格这的字符串写出去就好了,对吧,方便我们后续做处理啊,也方便输出。嗯,那接下来呢,我们就按照这个格式来一个一个的,那我们首先要获取数据库和表明,那怎么获取呢?很简单,我把这个拿过来,CTRLC。我定在这个。桌面上啊,这样的话我们就方便来看啊,这是我们的数据,也就是说它是它啊。诶,这个东西我刚才丢了,然后重新复制一下啊,然后呢,定在桌面上。呃,那我们看这个地方呢,Source record其实就是整个的当前这个对象,是不是当前这个对象它to方法打印出来的,OK,那这个对象当中呢,它有很多的属性,我们可以去拿,首先我们看到那我们的目标是库明和表明,那库明和表明呢,在我们的一个属性叫topic当中,大家不知道看没看到这个地方呢,有一个CDC test户名,这个地方呢是表明。
06:19
对吧,所以呢,很简单,第一个啊,库明跟表明在一块,我们获取库名。看到。表明很简单对不对,那怎么获取呢?呃,Result点,那我们看一看有没有topic这个属性。哎,并不是result拿错了,应该是source record对吧,点topic这个属性,诶刚好有topic这个属性,返回值呢,是一个string类型,那也就是说它这个整体这个就是我们的。Topic,那接下来呢,中间是用点分割的,所以我们做一个切割,取第二位,第三位就是我们的库明和表明,那这个就简单了,topic.display按照呃,空格分割啊,不是空格是点,点的话我们做一个转移。
07:04
对吧,这个转一二加V得到一个fell。好,那么接下来呢,我们就能够得到透明和表明了,我们把它放到数据,放到我们最终结果集当中,Result。点put啊,假如说叫DB啊,当然这个属K的名字,咱们随便取对吧,FALSES1好,那接下来呢,点good叫table。然后fails。二。二那此时呢,前两个我们就搞定了,对吧,那中间获取数据本身。好,那我们这样分为两块,Before数据。以及获取阿数据,其实它两个获取的方式都一样,只要你会一个了,那另外一个呢,也没有问题阿的数据。好,那我们看一下这个数据呢,应该在哪个属性当中,是不是来。
08:04
在哪个属性呢?呃,我们看一下before,还有after都在我们的叫value属性里边,好,那我们同样的我们的source record点要用这个value好不?呃,不像topic,它返回值是S类型,它呢返回值object,所以呢,我们要做一个转换,那因为object接下来我们没不好对这个value进行处理的,我们给它强转成什么呢?叫structure。啊,Struggle的类型,那这个struggle呢,注意导的是。卡夫卡连接器包下的这个内容,因为他往往卡夫卡这个来的啊,RWV得到一个value。那将来我们拿着这个内容呢,继续去get,诶,这个地方呢,就有字段名了,我们看一下这个地方就有一个比before对吧?Before,当然这个before呢,它本身也是一个struggle,我不知道大家发现没,所以呢,此时我们可以用这种方式来get,当然你可以这样啊,然后把它强转成structure也一样,当然这个地方呢,它可以直接get structure来得到一个before。
09:10
啊,Before,看得到一个before,那接下来呢,我们要解析这个before数据。啊解1BEFORE数据,那对于before数据而言呢,我们应该也同样的有一个切分对象来放到before里边对吧?所以很简单,我们先创建一个杰森object,然后呢,这个内容我们就叫。Son啊,Before的一个Jason,然后呢,我们往这里边去放我们具体的这个值,ID name这样的值对不对?好,那这个时候我们应该怎么去处理呢?对于before数据而言,我们想啊,Before数据有没有可能为那。那是有可能的,对不对,那比方说我们是一个新增数据。它只有after,那before数据是none,你直接对它处理肯定就空指针了,所以呢,我们做个判断,If before,它不等于none的时候,我们才能够做加工,做处理,接下来我们就是如何取出来before当中的数据,给它放到before杰森里边,那我们就拿这个内容来看一下点。
10:19
它也是get字段,但是这个字段并不像这个整体的字段,我们知道它叫value,因为这是每一张表当中的数据了。这个字段呢,并不清楚啊,所以呢,在前面我们要做一个什么事呢?获取before当中,我们先要获取它的一个。远数据信息。啊,我们要把当前这个原数据信息获取到啊,那我们可以通过这个Y6点。嗯,比方说我们来获取他的一个STEM信息,我们拿一下他的一个STEM信息啊。我们看一下在这个当中。STEM信息对吧,好看单位得到一个scam value的scam,当然如果说要获取这个before,那我们也可以拿着什么before来做一个获取。
11:05
得到它的一个STEM信息。啊,就是它的元素信息,我们就可以得到它的一个ID啊等等这些内容啊,所以呢,这个东西呢,我们还是给它拿进来啊,因为我们要判断它不等于到才能获取,然后接下来我们拿着这个STEM信息来看一下有什么东西点。哎,这个地方有一个false,刚好我们要就是它这个呢,其实就是我们的列名了。啊,其实我们的列名这个是获取列名,我们稍微写一下注释,获取列信息啊列名信息,那接下来呢,我们就对当前的这个list做一个循环。啊,这是一个一个的字段,那我们可以根据字段去获取值了,很简单,我们就before,杰森点put,那fail.name哎,列名有了,然后值值的话,我们从before当中去取,点get,把这个field传进去。
12:01
那我们值同样的也获取到了,同样获取到了,对不对,那接下来呢,我们可以把这个放到我们杰森里边了啊,Result点。我们的K呢,叫before。然后这个数据呢,是before杰森。好,那同理,我们获取阿的数据呢,一样的道理,只不过说这里边呢,我们要改一下这个Y6,我只要获取一次就够了,这个地方呢,我们获取的是阿。好,无非就把刚才的所有的before的地方改成after,这个地方注意大家在改的时候呢,千万千万不要漏了,对吧?好,这个地方呢,也是after,这是after啊,最后呢,这个地方是after介词啊,这个地方呢,我们改成after。好,那我们稍微观察一下这个东西呢,有没有。错误,After对吧?不要有before的内容,哎,这个地方错了,是不是这个地方呢?我们应该用al杰son来做这个事情啊,但ID要提示说我们这块代码重复度比较高啊,那因为我们写一个测试嘛,所以我就写一块了,当然在设置网件当中,我们可以把这块内容操作呢,可以提出去提出形成一个方法,那对于before和up的数据呢,我们都可以这样去处理好,那我们只剩最后一个字段呢,获取操作类型对不对?获取操作类型好,操作类型这个地方呢,我们看一下,它叫op。
13:24
Operator啊,那我们就想着呃点。Op。啊,这是option并不是的对吧,它这个选项好,那也就是说这个磁带获取呢,稍微要麻烦一点,跟其他的不一样,其他的我们直接获取到对吧,那这个呢,我们就直接从这个地方在文档当中我也给大家标红了。在这块我们可以把它拿过来。往那放对吧,导一下这个包好它呢叫lo,然后呢,Operation for,然后把我们的数据放进去就好了,然后呢,这是operation operation啊好,那将接下来呢,我们将它也放到结果题当中,without.food。
14:07
叫op,我们就叫op,然后呢是operation。啊,当然呢,你也可以拿它去调用一些其他的内容,对吧?啊,那它是一个枚举啊,来看它里面呢,有。CD啊,如果说呢,还有一个清空表传K对吧,当然我们更多的刚才测的是这个。对于表的数据操作,DML没并没有测这个。创建表啊,删除表啊,这种操作对吧?好,那这个我们就写完了。啊,写完好以后呢,在弗林和CC2当中,这个地方我们就不需要去拗自带的阵容,我们应该六叫customer。Ization的一个STEM信息,对吧,是这个意思,好,那下来呢,我们直接运行一下,看这个结果还是初始化的啊,那这个地方呢,我们还是有很多的数据的,对吧,这是两条,嗯,这个呢,也有六条数据,八条数据,我们稍微看一下这个结果。
15:09
我们写这个自定义的反驱装器呢,是为了后续对数据加工更为方便一些。好来看,在这个当中呢,我们看到before都是为空,因为它都是read啊,它是第一次初始化的时候,它去select查询我们整张表的数据,接下来呢,是up的数据啊,我们有DB数据库,有表,表呢有user和user for2,那此时呢,比方说我做一个修改,把这个ma呢改成非妙。保存数据。那此时看是不是before跟after都会有数据。啊,Before跟after都会有数据的啊,那再比方说我们删除一个数据。删除。那删除数据呢,Before有数据,After没有数据对吧?所以两个呢,我们都做一个保留啊,当然这两个表呢,Use in for表,我们再换一张表做测试。
16:03
U2IN2的这张表,我们再把这个CC呢改成ddd啊做一个修改保存,那此时呢,看表明是UC2。对吧,然后呢,也同样带了before和啊的数据,那这个时候就完成了,那你说我们现在把数据变成这个样子,是不是未来操作会更方便一些。如果说用默认的它这种方式,那很明显太不好解析了,对吧?好,那这个呢,我们就自定义的反吸化器,到这就结束了。
我来说两句