00:01
OK啊呃,那接下来的话,我们就先不考虑那么多了哈,这个先不考虑什么各种什么驱虫的细节呀,这细节那细节的啊,你先把最基本的功能做完。就你先能保证把数据给它消费出来,对吧,说下面我们写的就是消费数据的这个过程了啊来过来,呃过来以后我们在这个APP里面再去写上一个类啊,这个我们叫做DWD层的啊DWD层,然后呢,日活叫dau是吧,Dau的一个APP。OK吧,就叫这个名字了啊。好,呃,那么这个类我们主要做什么事情呢?我们主要做的就是这个出日活宽表啊。好,任何宽表行,呃,那么按照我们刚刚的分析啊,我们其实也可以把这个整个要处理的事情呢,给他罗列出来啊,第一步。那你还要什么通过这个Spark streaming去消费数据,对不对,所以说你还得去准备什么,准备你的这个实施环境啊,准备实施环境对吧,那你准备好以后,你想想啊,我还是从你的卡不卡去拿数据,那你要从卡巴拿数据,你作为一个消费者来讲,我就得考虑漏消费,重复消费问题。
01:15
对吧,只要你想到这个事情,你你就知道我们是不是可以依旧使用我们之前设计好的一套。管理方案了吧,对吧,还是啊,我先保证什么,先保证你这个数据呢,先不丢失啊,你重复你就重复呗,对吧,重复就重复,反正我最后再往这个ES写的时候,往我的olp里面去写的时候,我是能够保证你这个精确一次的。对吧,然后去重不就完了吗?所以说你第二步还是什么呀,我们要从你的red中哎,读取什么读取这个偏移量好第三步然后呢,从卡不卡中消费数据啊,好,然后呢,第四步我们要去提取偏移量的什么这个结束点对吧?好然后第五个的话就是我们这个处理数据了啊,那这个处理数据的话呢,呃,我们应该也能想到啊,首先你会有这个什么数据结构的转换啊。
02:10
转换数据结构,因为你拿出来以后呢,它还是一个什么呀,这个这个叫什么叫这个,呃,拿出来以后,它就是一个什么消息,对不对,那你这个消息你还不能够直接去用,你要什么把它里面最后的最终的那个什么,就是那个Y6给他拿出来,对吧,那拿出来以后,至于你要处理成什么结构啊,你到时候再来分析呗,啊,什么通用结构呀,什么专用结构啊,对吧,都可以啊。好,那处理完成结构以后的话,下面就是我们的驱虫操作呗,刚刚分析过的啊驱虫然后做什么做这个维度关联对吧,你把这一套都做完成以后,那我们最后就可以把数据呢,写入到你这个ES对吧,那这个写入的话呢,我们就呃写入的话呢,就单独来去说吧,来这个第六啊就是这个写入。啊,写入这个ES。
03:00
好吧,好,这个我们暂时还做不了啊好,那等你这个写完以后的话呢,你你是不是就可以考虑那个啥了呀,叫这个提交off对吧。啊,这一次的话呢,我们就不往这个卡不卡写了啊,所以说那个卡不卡那个缓冲区那个问题呢,呃,也就不存在了。明白吧,啊,也就不存在了啊,那这个ES有没有这个问题呢?我们到时候讲了以后呢啊,再来去分析呗。好吧,再来去分析啊,行好,这是我们这个完整的一个处理流程啊,然后下面我们就通过代码呢,一步一步把它实现出来啊,我们先来写卖方法,好这个第一步,呃,我就不解释了哈,直接写了哈,其实这个代码我们直接拷贝也行,对不对,好我还是写写吧。好,来这个set一个APP name我们就叫DWD,然后呢,Dau的一个APP吧,好set一个MASTER4啊。
04:04
Spark OK,呃,Spark对象啊好,然后呢,再去用上一个streaming contact,好把这个SPA com传进去,然后呢。穿上一个采集周期啊,比如说还是五秒钟啊好OK。行,那积分的环境都准备好了啊,准备好以后呢,接下来第二步就是,呃,还是什么呀,从red中读取偏移量啊,偏一辆,OK,呃,不是便宜啊,就读取outside吧,好吧,读取outside得了啊,行,那这个怎么读呢?还是使用我们的MY啊,这个叫opposite,叫什么叫做read?好,那你读的时候我们需要用你的topic,需要用你的这个主ID啊,那我们就定一下吧,呃,Topic name。
05:06
对吧,行,那你说我们这一次从哪个topic出数据呢。刚刚我们分析过啊,我们的日活任务,我们是从什么呀,从你的页面访问来出数据的,所以说呢,你就要知道你的页面访问这个topic叫什么名字,那这个你得看谁呀,你是不是得看你的第一层就是你的base log,你再去分流你这个页面数据的时候,我们把它写到哪个topic里面来,我们的。Page是不是这个名字呀?对吧,所以你要前后统一的啊,你前后不同意你接不起来的啊,行,再来写一个呃group ID啊好,这个我们就叫做呃。哎呀,这个我叫啥,我叫这个DWD,然后呢,这个。好吧,Dau的这个一个group吧,OK吧,我就不再写那么长了哈,麻烦啊,我就这么简单一点啊,行好,那把这写好以后呢,我们就把这个topic name传进去,再把这个GOID传进去,那我就会读到一个。
06:13
OK,拿到这个off以后呢,下面第三件事情,那就是从你的卡普卡中消费数据了吧,对吧,消费数据啊,OK,呃,那就写一下吧,还是先判断啊,你的opposite是不等于空的,并且啊,这个opposite的长度就是这个叫什么叫now empty啊好,这种情况下我就可以去读了啊叫麦卡卡有跳点。Get,卡不卡?Stream,把SC传进去,把你的topic name传进去,把你的GOID传进去,把你的opposite传进去,好,这叫卡夫卡stream,那我把它踢出来。呃。提出来啊。
07:00
好,等于空啊,然后再给它赋个值啊。能看到吧,这样好,那如果说哎你是不满足条件的,那我在读的时候呢,我就不传set。对吧,这就把数据读到了啊,读到以后呢,下一步操作呢,就是第四步啊,我们需要提取这个偏移量结束点啊,提取ET这个结束点啊。怎么提呢,使用卡不卡stream,然后做一个transform,你给我传一个RDD过来,然后呢,我就s instance of,然后呢,我们叫做has,呃,Opposite ranges对吧,然后呢,去调一个方法好接收回来。把这个还是提出去啊。好,改成VR的,要不我们不停能重新赋值对吧,拿回来放到这OK,最后呢,把RDD返回回去啊,行,那接收回来,这就是我们的叫做opposite bridge的一个stream啊。
08:14
可以了吧,行,那提出来以后,接下来下一步呢,第五步啊,就是对这个数据的处理了啊处理数据那首先第一步就是什么转换结构了呗,转换结构啊结构OK,呃,那你要转结构的话,我们就是什么直接做一个map操作。是不是?那你map的时候大家想想啊,他给我传过来还是一个什么呀,还是一个这个什么consumer record对吧,那我们就还是consumer record OK,那我把它转成什么呀。我们之前呢,都是把它什么,把这个呢,先把它value踢出来啊,我们主要是看value啊,这是我们的这个叫做,呃,就叫吧。
09:01
好,那之前的话,我们都是把它转成一个什么摘对象。是不是啊,这是一个通用的一个对象啊,呃,这一次的话呢,你你自己看吧,你是想把它转成这个战胜对象呢,还是说想把它转成这个,呃,就是我们的专用对象都可以,因为我们是有专用对象的啊,你看一下我们之前在上一层做分流的时候,当我们拿到一个什么页面数据以后,你看一下啊。我们看我这个分流代码,呃,如果说你是页面数据的话,我是把它封装成了一个配置log,然后呢,把它写过去的吧,那就说白了,我在写入到它里面的时候呢,其实我是有一个配置对象的,我把它配置lo呢,转成了一个什么再生字符串给它写进去了,那现在我拿出来以后呢,就是一个再生字符串,我是不是可以再把它转成一个配置log呀。对吧,既然你你是有专用对象的,那我就可以把它转成这个专用对象啊,所以说把它转一下了啊,怎么转呢,还是Jason啊呃Jason叫叫帕呃先打爆哈。
10:02
好,Pass object,然后呢,把你的value放进去,注意后面我们需要再传一个值啊。嗯,后面传什么,传一个就是你的type,就是你的class类型。能明白吧,好,那我要转成什么,转成这个class off,我要让它转成配置log对象。能看到吧,就告诉他,你把这个加成字符串给我转成这个配置log对象,好接收回来,这是我们的配置log,那我最后呢,再把这个配置log呢,作为一个返回。好,这个转完以后,我们接收回来叫配置log的一个stream。OK吧,行,那我们就先处理到这啊,处理到这以后呢,其实我们就可以什么先去测试一下,看看我到底能不能真正的把数据呢,给他这个消费出来,如果说能够消费出来以后呢,我们再去考虑做这个驱动操作,好吧,来我们走一下啊呃,别打这么多了,100行够了啊行启动起来啊。
11:00
呃,那你这个启动起来以后想想啊,嗯。稍等啊。这一次我们得配合着用了哈,配合着用了。好,来看一下啊呃,现在我们写的是,现在我们写的是这一层啊,同学们,我们写的是这一层。对吧,那我们从你的这个topic中呢去拿数据,那你这里面的数据你是怎么来的呢?你这里面的数据是不是我的上一层写进来的呀,同学们。对吧,是我上一次写进去的,然后呢,现在我要什么再把它拿出来,所以说你的这一层如果说想看到这个数据的话,那你的上一层是不是也得启动起来,然后呢,开始往它里面放数据啊。对吧,所以说把这个配合起来了啊,同学们把它配合起来来,那我就把它先扔到这儿啊,配合起来,呃,这是我们的下游啊,我把它这个清理掉,然后呢,接下来我们再把上一层砌起来,上一层是我们的,呃,Base logg,对吧,软起来。
12:07
对不对,这是我们的日志数据的采集分流啊。好了,呃,这个起来以后呢,接下来我们就可以生成一波数据了啊,我们来看一下呃。这个我我退出啊,诶好我退出啊,来我们上传数据啊,那我们的脚本叫Lg.SH那我就写个2022年杠零三杠二二好走。这就开始什么生成数据了,那你这个生成数据以后呢,我的呃,直发的啊,我是直发给卡不卡,然后呢,我的上一层呢,就会把数据做一个分流,分到你的这个主题中,那你分到主题中以后呢,我在下一层就可以什么从你的主题中拿出来数据,然后呢,通过我的Spark stream呢,把它什么消费出来,对吧?所以这个时候你就是什么可以回到你的控制台看一下。
13:06
有数据了吧。对吧,就有数据了啊,我们现在拿到的就是你的什么配置log对象。能不能看明白啊,现在我就有数据了啊,OK吧,同学们行好,那你能够看到这个效果的话,就说明什么,说明我们的,诶这个消费数据是没啥问题的,那接下来你就可以什么继续往后去处理了。OK吧,停一下。
我来说两句