00:00
呃,那接下来呢,第一个获取这个执行环境,那这个呢,我们就直接还是一样的去拿一个啊。把这个整体拿过来就行了。没必要每次都去敲对吧,这是第一个,第二个读取这个卡夫卡主题的数据,然后呢,创建这个流对吧,那就因为点ADD source叫MYKA u.get卡的一个consumer。呃,然后这里边呢,有两个东西啊,第一个呢是string叫S。To。然后我们赋值就是它叫ods倍CD。下来第二个。叫这个group ID啊,Group ID呢,咱们用这个。
01:00
啊叫贝DB ABB,然后呢,小发现0210325啊这个消费者组我肯定没有用过啊好,那接下来呢,这个地方就放这个source topic和啊那这个呢是比较简单的,我们就得到了一个卡夫卡的一个点。好,得到了一个卡法DS,好,那接下来我们将每行数据转换为杰森对象,并过滤掉这种delete的数据,对吧?它呢,我们就不要了啊,它就不要了啊,OK,那我们先转换一下,我们就分两步写了点map,这个地方我们可以用map,因为前面的阶层是我们自己封装的啊,它并不会产生那种什么脏数据啊,我们自己封装的啊,但是你要不要担心啊,那你也可以加一个这个process做一个过滤,对吧?啊,把这个脏数据过滤出来也可以,但一般这个时候没有脏数据,业务数据嘛,它不像行为数据对吧?业务数据当中呢,我们一般没有脏数据,那除非那业务数据当中脏数据呢,特殊字段。
02:01
对吧,为nu,那这种呢,你跟我们现在讨论的脏数据不一样啊,我们是完全谁都用不了的这种数据,对吧,那你那个是特殊值这种过滤,当然你可以做一下,我们这边不做了啊,那这个是赖这个地方呢,就是杰森。点pass。Object,然后呢,这个赖进去。当然这块有简写的方式对吧?啊这样去写啊,然后点过滤。啊,做一个filter啊,那我们有一个方式。呃,那这个是value是我们的介object的,我要把这个delete的数据给它干掉,对吧,所以呢,我们先取出它的类型。取出数据的操作类型,好,那就Y6点,大家记得吗?在get string这个里面呢,有一个T。对吧,注意啊,这个东西呢,操作类型要根据我们方程当中自定义的反序列化这个地方来,对吧,我们封装的数据呢,叫看。
03:07
我们主要拿它对吧,好到这边来看好单位。那是定义。我们就。不要了,对吧,如果你是一个定力的数据,那我们就直接过滤掉啊,直接过滤掉好,所以呢,这块当然可以用三元运算符,你当然也可以这样写一下,它有个自动生成啊,那比方说delete的。变equals这个谁呀,Type啊,那如果是delete的,那怎么办,我们返回。这条数据就不要了,否则返回什么。数啊,否则返回数,当然这边呢,他告诉我们可以用三运算符。啊,直接返回它就好了,对吧,取反啊,就看你是不是给力的啊,就取反就好了啊行,那这个呢,就过滤掉了啊抗R化V得到了一个杰森。OBJ的一个D。啊,得到了一个杰森object的一个DS对吧,这个相对来说比较简单一点,当然有同学可能想,哎,跟文档当中可能不一样啊,那这块呢,你也可以多加一个过滤,因为我们如果是拿到了这个,呃,新增或者说这个更新数据,假如说他的after数据没有多少,那你也可以给它过滤掉。
04:22
对吧,它的after数据没有多少也可以过滤掉啊,这个没问题,那你直接用定delete的过滤掉也可以啊,也可以就是这个意思啊,就是你可以直接拿after来做过滤啊,比方说这块还可以怎么写啊,就是文档当中我们是另外一种写法。这个。啊,当然我以前风装叫得塔,这个得塔就是我们现在写的after对吧阿,它不等于呢,并且它的一个length大于零啊,其实我们应该大于三啊,等于三啊才可以给他留着,那其实我们就是过滤掉这个滴类的数据,直接这样写更简单一点对吧?好,那这个就是消费并过滤数据,这个呢,比较简单啊,我把这个解一下下来就。
我来说两句