54.腾讯云EMR-实时数仓搭建-DIM层-处理连接流-主流-读取状态&过滤数据原创

呃，那接下来呢，我们应该处理这个。主流数据对吧，主流数据呢，同样的三件事情，诶，那我把这个注释先写上啊，第一个叫获取广播数据对吧。第二个过滤。数据这个分行和列两个层面吧，两个层面的过滤对吧？好，那第三件事情就是补充think。Table。字段。写出。对吧，啊，那就是做这个事情好，那接下来呢，我们来写第一个获取广播数据，这个呢好写对吧，那首先呢ctx点。Get啊，然后呢，还是一样的。呃，描述器放进来，这个就得到了一个read only，对吧？只读的广播变量，然后呢，在这个广播状态当中要我们的数据，当然呢，我们获取数据的时候点get，是不是要传一个K啊，这个K呢，是表明对吧？那这个表明应该怎么来呢？是不是在这个value里边。

对吧，所以我们得搞清楚这个value它。这个数据又是什么格式对不对？没毛病吧，好，那你想这个数据。又是怎么来的？它的格式我应该找谁？对吧，它的格式我应该找谁。诶，刚才提到了mawelll，对了，它是mawelll格式吧，它是macel格式，好，那我去拿一个啊，Mawell的格式，其实我这边有，我这边有一个。

Li CDC的格式，诶，长这样对吧？呃，然后这个是Maxwell的数据格式，这是普通的日志格式啊，我这边其实都放了，呃，Mawell封装数据呢，长这样子，嗯，我呢得这样什么啊。我再来一份啊，我干什么事儿呢，我这样。我得把它呀放到一行啊，我我这个自己加工一下，我快速处理一下好吧。我先把这个，诶错了。把那个干掉啊。嗯。

把它。缩进一下啊。就把它变成一行上去，因为这样我们拿到我们的代码里边好拿一点啊。这是我们的格式，你要知道这个表明它等会儿在哪对吧，你要看一下啊。好，那我们的格式呢，就长这样子对吧？啊，直接拿过来啊，这是我们的格式啊好，那我们要的表明这里面是不是传表明啊，那表明在哪呢？来看一下啊，直接看这个地方也行啊，叫database table，诶就这个吧，直接在大的阶里边叫table，那就简单了对吧？好，那这个格式，然后接下来呢，在这里面写Y6点。Get，诶，就叫。Table。对吧，好，CTRL加V得到一个table。啊，得到一个这个叫table也好，呃，Table process啊。

叫吧，更好一点啊。好，那接下来呢，我们要做这个过滤啊，那我们先想啊呃。这个东西。这个表啊，你比如说这边我来了一条数据是cut for，那你说我拿着这个cut in到这个广播状态里边去取数据。他取出来的是个什么东西啊？你就想在我们的配置表里边，你会不会写一个东西叫cut info？看的衣服购物车对吧。购物车，它肯定是属于这个叫。事实表。对吧。

那。他应该为none吧，对吧，如果是事实表，就是不是你所需要的表，它都是none啊，比如说这边呢，总共有46种，46张表的数据的可能性，但是呢，广播状态里边只有十张表，那是不是有36张表获取的都是那么所以我们首先过滤，诶那怎么做判断。If。Table process，它不等于那那才是我们要的这个数据。对吧，我们在里边再去过滤列吧，那除了这个数据之外，我们还有没有数据不要。还有。还有要过滤的啊，那我们想一下啊。呃。通道删除数据怎么办？比如说。嗯，这样给大家说啊来。

比如说base trademark这张表，首先呢，在我们配置信息里边啊，我举个例子啊，呃，这个数据呢，我先删掉啊，假如说呢，有这样一行数据啊，Source叫base。Trademark。诶。咋回事？刚才是敲了什么东西了？好，比如说这个是DM小划键被缺mark啊c columns，然后一个ID逗号隔开对吧，一个叫TM name，呃，主键呢是ID，这个呢可以不写了，保存对吧，比如说我有这样的一张表，表明我要这个数据，对吧？好正常的，你往这张表里面新增啊，或者更新啊，那没关系，我直接往菲去写，它会自动更新，好那如果说呢，我在这里边删了一条数据。那你说这个删除的数据你得想啊，对吧？呃，因为如果是删除数据的话，我们看这个格式啊。

他的data你要写出去吗？对吧，那你要碰到删除数据，那其实删除数据咱们是不是也可以不处理。删除数据是不是可以不处理？对吧，那我们想如果不处理它会发生什么问题，好，那比如说。这边有十条数据啊，买搜狗。有十条数据，好，那我第一次呢，先同步过来了，在Phoenix里边也是十条，好，这边呢我删了一条。啊，此时呢，变成了九条。那对这个删除数据呢，我不做任何处理，我这边不管对吧？啊，我直接给它过滤掉，这边还是失掉，那你想这边数据不是不一致吗？它会不会有问题呢？虽然我们说MYSO跟这个Phoenix数据不一致，但是你要思考一下它会不会有问题。

想一想，你们觉得会不会有问题，如果说有问题，咱们得把这种删除数据拿到，对吧。嗯嗯。其实这种情况下没有问题，为什么呢？你想啊，在MY里边只有九条数据。那未来我们生成业务数据的时候，是不是只可能在这九条数据里边去生成？

是不是只有可能在这九条数据里边去生成啊。对不对，只有可能在这个里边去生成，好，那也就是说无非是这边呢，我多了一个第十条数据，但是呢，未来这第十条数据是不是没有人访问。是不是应该就没有人访问，所以咱们应该把这个什么，把这个删除数据，如果它delete对吧，我们也直接过滤掉，能不能明白。如果是删除数据，我们也不做处理。尽管你的数据。是为表数据，但是你是为表的删除数据，我可以不做处理，这块能不能明白，能明白给我扣个一好吗？

OK吧，好，呃，那除了删除数据咱们不要，咱们还有一种，因为我们之前提到了，现在呢，在我们的表里边儿已经有数据了，所以呃，第一次我们为了能关联到数据里边，我们是不是应该把这个数据整个的都导到Phoenix里边，对吧？把这个历史数据都倒过来，好，那历史数据倒过来呢，我们要用。MY的boostop的功能吗？不，Top的功能，它头尾两条数据大家还记得吗？它是处于脏数据对不对？头尾两条数据是不是属于脏数据啊？还记得吗？

在我们做这个初始化功能的时候，它头尾是不是有两条脏数据对吧？头第一条数据叫put start。对吧，最后叫bootop的complete，但是这里面都没有数据，它属于脏数据，它表示我们初始化开始和结束。一头一尾是不是这两这两条数据是不是也要给它过滤掉，对吧，你得考虑清楚你有哪些数据对吧？好，这个大家可能忘了，但没关系，我给你测一下好吧。呃，来。打开。那我们要打开这个ZK卡不卡了，对吧，我们待着你测一下啊。卡布卡。好，然后。好吧，那这边呢，我们起一个消费者先啊。

呃，咱们呢是topic DB对吧，诶去消费这个主题，接下来呢，我们开启Maxwell啊。嗯。呃，咱们要做它的一个初始化啊，这边呢，我们再开一个，我去拿一下这个初始化的命令，好吧，啊，其实大家呢，都有这个脚本，因为你们之前的离线收仓导过这个全量的实时表，对吧。

呃，我去拿一个就是它啊B把它，然后呢，把它拿过来，我改吧改吧啊。好，那数据库呢，咱们用这个220212这个数据库对吧？呃。麦刚刚带着，然后呢，表明咱们用base。真的吗？好，那这个呢，我们就mal home就点杠对吧，我们到到这里边去执行嘛，对吧？好，那我们做一个初始化给你看一下啊。在这边对吧，走你。好，这边呢，开始结束了，我们来看这它有同为两条脏数据啊，那第一条数据它叫booto start。看见没这个对吧，好，那么还有。Complete这两条数据呢，都不要，因为它data是什么空，那没有用，这只是头尾的一个标记，所以咱们要的是什么数据啊。

是Photoshop音色的，要的有三类新增更新和这个Photoshop的音色的对吧？啊，那所以呢，咱们呢，把这个写一下啊呃，很简单到这边来。并且对吧，括号诶那类型呢，我们得获取一下，把这个类型获取到对吧。Y6.getstem，哎，就叫TQ号加V得到一个类型来看一下是不是啊，在这个地方有实例数据。叫type吧类型啊好，那这边呢，我们直接在这儿来写啊。银色的。点。看。啊。或者。啊，对的。点equals咱们的in。对吧，啊或者谁。不算。

不是算音色的，诶这个呢，我们也是需要的点equals咱们的什么。Table对吧，啊，那他们三者肯定是或者的关系，对吧，三其中之一都可以跟他呢，肯定是并的关系啊，你必须你不等于那对吧，且是你才可以，这是我们要的数据，这个其实做的是什么呢？行过滤，整行数据做过滤。对吧，好，那else。这个是。不必要的数据对吧，我们打印句话啊，或者你不打印也行啊，打印句话呢，你未来有很多了，因为咱们实时表未来还是更多一些的，我就不打了，对吧？啊大家知道一下好。咱们这里面别着急，这里面呢，还有一个列过滤。是不是咱们还有一个列过滤，现在呢，咱们写了这个判断只是行过滤，整行数据我不要啊，那还有一个列过滤啊，因为我们说了你的数据呢，可能是这样子的，Idtm name logo ul，但是呢，我要的数据只有ID跟TM name，那建表就两个字段，你未来的数据里面是三个字段，那不行，我要把这个字段过滤掉，对吧？这是列过滤，把这一列过滤掉对吧？好，那这个呢，我们叫filter。

卡了。呃，那传什么东西进来呢？那必然要把这个data传进去，对吧，数据里边这个啊，数据本身传进去data对吧？呃，然后还有呢，咱们的这个table process里边的。Columns对吧，这两个字段就好了，那这边呢，就是value点。Get接生object，因为要加工，所以呢这样写啊date还有一个呢，是我们的字段table process.get s columns对吧？这两个字段，诶我们去做一个过滤，那就out加回车，创建过滤的方法。啊，好。

过滤数据。啊，它呢是列啊这个上面呢，其实是行啊。对吧，这是过滤的列的数据，好，这是date，这是s columns，好，那我们想这个其实比较简单，那杰森，那我们把这个里边列呢都取出来。对吧，便利啊，便利在这里边判断看在这里边是否存在，如果存在了，那我就要，如果不存在我就不要，对吧？啊，那按照这个思路来写啊，呃，这是杰森，那怎么对它进行便利呢？其实你要注意杰层object未来遇到了，如果说你要对它进行加工或者便利什么事你就当做map来处理好，那我问一下大家。对于map这种数据结构，我们可以怎么遍历？啊，Map数据结构怎么遍历啊。

Map这个数据结构怎么遍历？ENT，哎，可以，还有没有。想一想啊，就是趁着这个时间可以大家回顾一下嘛，对吧。转化为ENT set基变利还有没有可以，还可以怎么便利？Key set，对啊，还有key set对吧？啊，我们这边呢，肯定用n set进行编，因为我们要整行数据就都不要了啊，KV都不要了对吧，那我们就。Data点。Entry set啊卡尔加V得到一个entries对吧，好，Entries点。

变成一个。迭代器。单位对吧，得到一个叠加器，好，那这边呢，Y。叠加器点has next，然后呢点next。对吧，得到了一个鹌。呃，那我判断。If。这个s columns点。Contents。啊，如果包含next.get key，对吧，包含我们就需要，那我们要做的什么呢？不包含去反对吧，如果你不包含，那我就移除掉。好，那怎么移除呢，那就点。为目就好了，对吧，直接把它移除掉啊，就这样写一下对吧，好呃，那这样写的话，它有没有问题啊，有没有可能有问题啊，大家想。

这里边儿其实有问题啊，我既然这样问了，对吧，肯定有问题，但是就是你们能不能发现出来这里边有啥问题。Think column转成list three类型的。对了啊，其实问题就在于这。啊呃，这里边有什么问题呢？可能大家还是看不出来，我给你举个例子就知道了，比如说咱们的数据啊，Data。贝塔呢，它里边这样子的啊，我就不具体写了，我就给你写字段好吧，Value我就不写了，对吧，比如说呢，里边有ID有name，有TM name。对吧，有三个字段啊，有logo URL吧，比如说啊logo URL对吧，诶我多加了一个字段，你看见了吧，对吧，叫name啊好，咱们这个think columns这边呢，它有这几个字段。

诶，它是一个字符串对吧？呃，有ID有TM name。诶，就这两个，你想想看有没有问题啊。本来我是不是只需要这两个字段呢，但是。但是怎么样？他存不存在存在。这个在里边有没有有，因为它有包含关系。它有包含关系对吧？啊TM name呢，存在存在啊，那防止这个字段与字段之间有包含关系，你呢，不要最好不要直接用spring，而是把它转成。集合对吧，好，那think column点按照。逗号隔开CTRL加V，得到一个columns。对吧，但是呢，这是一个数组，数组可没有什么content方法，所以刚才有同学说了要把它转成集合，对吧？啊，那怎么做呢？那就用。

点at least，把这个放进去对吧，卡尔加V得到一个column。Least。对吧，好，那你如果是集合的话，你就有。方法了。对吧，啊，转成集合就不会有，刚才我们这个问题，这个问题大家能看明白吧，应该不难明白吧，这个那你看啊，S columns里面有两个字段，本来我需要两个字段，但是你这个name在如果直接用字符传。是不是也能保留了？就多保留一个不对的吧，这个能明白吗？给我扣个一好吗？我发现怎么周一大家比较蔫，到了周二还是这样呢？兴奋一点啊，我进班的时候就听说咱们班比较活跃啊。啊，你这哪活跃了，我真看不出来，还没有之前人家30个人的班活跃呢。

咱们是70个人的班。我前面两个班带的是30多个人的班，你还没人家活跃呢，对吧。这个能明白吧，啊，大家动起来，要不然你就容易犯困，你知道吧，就昨天有同学说上课就在那发呆。对吧，啊，你跟着思路跟上，然后呢，叫你发弹幕的时候敲一敲活动活动也会能好点吧。对吧，跟你们说的都不信啊，就是咋说咋不信，咋说咋不信啊，那你找工作的时候就不要问我了好吗。反正说了你也不信对吧，然后找工作的时候这问那问。好，那这个呢，咱们就搞定了啊，就是迟到多的同学你记住啊，在我这边拉入黑名单了，你知道吧。好，那这边呢，飘黄，因为它有一个简写啊，就是可以用简写的方式，那怎么做呢？来这样我把这个整体。

复制一份放在这儿，这个呢我注释掉。这个呢，我注释掉好吧，然后呢，Alt加回车注意看啊。其实他俩一样的，他ID要给我们自动生成的，对吧，我们可能自己写写不出来这玩意，但是呢，就看得懂，因为咱们可是学过scalela的，对吧，好叫。Remove if对吧？啊，Remove什么时候删除呢？叫next便利啊呃，然后呢，这个包含这不就有核心条件吗？对吧？啊，这个应该能看得懂。OK吧，那那到这个为止呢，咱们过滤就搞定了，这是上面的行过滤，这个呢是列过滤对吧，两个过滤啊。