00:01
好,下面我们就通过这个代码来去实现啊,那代码的话我们就暂时先不看了啊,我们先回到。回到我们这个开发工具中啊,我们来以后先去写啊。呃,我把这个先停掉啊,这是我们的日活的啊,把它停掉。好来,我们再去写一个APP啊,用上一个scar啊,叫这个还是DWD的啊,我们就叫这个order吧,Order的一个APP。好吧,就叫order的一个APP啊,来加进去。好,写个注释啊,这是我们的订单快满任务啊好,呃,那我们还是这样啊,先去这个整体梳理一下我们要做什么事啊,嗯,第一步还是准备你的实事环境啊,对吧,那你有了这个环境以后呢,接下来我们就可以去读什么呀,你因为你还是从你的这个DWD去出的数据,说白了我还是去消费你卡不卡的数据,那么只要你是消费卡不卡的数据,那你就少不了要考虑到你的重复数据或者什么录数据这个问题啊,因此呢,还是使用我们之前的一套方案啊,我们要从你的red中啊读取这个offset,对吧?第三步就是什么呀,从卡夫卡中消费数据。
01:21
对吧,充卡吧,消费数据啊好消费到数据以后呢,接下来就是我们要什么提取偏移量的一个什么结束点啊,偏提取这个offset吧,对吧,提到offset以后,接下来就是我们的处理数据了啊,就是数据处理好,那数据处理的话呢,我们就是两步,第一步还是这个维度关联啊。好维度关联,然后这个第二步的话,就是这个双流的一个招引啊,OK,那等你把这个数据都招引好以后呢,接下来我们就可以什么写入到这个就是ES对吧,然后第七步的话就是你可以那个就提交offset啊。
02:01
对吧,还是你看还是这么几步啊。是不是啊,就是你的每一层的处理啊,就是你这个ods层的处理,那基本上都是通的,然后DWD的处理,基本上这个流程呢,也都是一样的。好吧,行,呃,但这里面我们需要这个再去细细的去想一下啊呃,以前的话我们这么一写,然后你就什么去实战代码就行,但这一次的话,我们会稍微会比较麻烦一点。麻烦在什么地方呢?因为是这样子的啊,我们从中读取opposite应该得乘以二。对吧,为什么乘以二呢?因为你看啊,这一次我再从你的卡不卡消费数据的时候,我是要消费你的两个主题的。那你的两个主题,那我不得分别去维护这个upset吗?对吧,一个是你的order info,一个是你的order detail。能想明白吗?同学们好,那同样你从卡夫卡消费数据呢,也是两份,那你提取off呢,也是两份。对吧,这个处理数据的话呢,就是单独对待了啊,你维度关联,我们只需要对你的order info做维度关联就行了,因为你的order detail里面其实是没有什么用户信息的啊,它里面只记录了什么,只记我这个订单的一个ID。
03:08
对吧,好,那当然它里面也能做维度关联啊,它关联什么呀,比如说你不是一个详情吗?那你out底消是一个详情的话,它里面就会有你的SKU的信息,对吧,或者什么SKU的信息。是不是这些东西你也可以去关联啊,如果说你想用到这SQ里面更加详细的信息,如果它里面有的字段没有的话,那你可以去做关联。对吧,这就看你这个后面要不要这个数据,如果要你要去关联,如果不要的话,那也无所谓。OK啊,这个一会儿我们去闲的时候再来去看啊。好,那最后这个写入ES,提交offset,那也是乘以二。对吧,所以这里面我们要什么都做两份啊做两份。好吧,来,那把这个说明白以后,接下来我们就先把这个最基本的功能写出来啊,一直写到这个。就处理数据吧。对不对啊,先写到这个能够把数据呢,正常的这个打印出来啊呃,那再再加一步吧,啊,这少一步啊,你得什么转换结构啊,对吧,你的转换结构的哈,是不是你结构不转换你就不好去处理啊,因为你读出来的数据它还是一个加成字符串。
04:14
那你要去转换结构,你是转成这个通用的,还是转成这个专用的。是吧,这个我们都很熟了啊行,那下面我们就开始写了啊,最后再写一次吧,啊,这个再写一次就结束了啊,后面就不用再去写这种代码了啊好,那首先我们还是用上一个Spark com OK,然后呢,先设置个名字哈,这是DWD的一个叫的一个APP。好像这种代码啊,有的同学呢,他就是说,哎呀,我都写了很多次了,是吧,我就不写了吧。我就直接去做一个复制啊,这个复制这个事情吧,我倒不是说这个特别的。就是说这个不同意啊,或者怎么着的,你可以复制啊,但是你复制的前提是你一定要很熟以后你再去复制对吧,你别这个自己写都写不明白了,你就什么直接去复制了。
05:09
不能眼高手低啊,这个你们现在不会写代码,有很大一部分原因就是代码量写的还是很少啊。就看到这个代码就犯怵啊,照着抄还行,你要让我自己写,根本就不知道从何下手。对吧?你们有没有平常想过自己的这个问题啊?为什么不会写代码啊?是这个代码有多难吗?好像也不是吧。想来这是我们第一步啊,准备这个环境啊,好,然后这个第二步,呃,第二步是我们就要什么从这个red读up了啊来这个从red中啊,就读取吧opposites啊呃,那这个读的话呢,我们就什么直接什么my oppositeet好讲这个read。
06:00
对吧,Read read的时候呢,我们有一个topic的名字,有一个god,那这个需要去定义一下啊,那我们就叫叫order in for topic name好吧,呃,那这个主题的话呢,你要注意啊,我们用的是哪一个。我们之前来看上一层了啊,同学们,上一层是我们的贝斯DD啊呃,我们的事实数据往这个,我们的事实数据往里面写的时候怎么写的呢?我们是拿着你的DWD,后面是你的表的名字,在后面是你的操作类型对吧,操作类型后面是你的这个,我把它拿过来吧啊。好,把这拿过来,拿过来以后你去改一改好那这个表的名字,因为我们现在是很明确知道读哪个表的哈,就直接写了order info,后来这个操作类型的话呢,我们目前的话呢,也没有什么修改啥的,我们都是一个什么新增啊,说说什么直接去用它了,那我这个需求的一个统计就直接使用这个。Insert啊,因为我们都是把它分开的,什么inser呀,什么update呀,什么delete呀,对吧,都是分开的啊,这个像我们目前的话呢,呃,涉及不到这些东西啊,就这么直接什么insert就完事了啊好,这是我的一个主题的名字啊,然后呢,Order in for的一个group啊,那我就叫什么就叫DWD,然后呢,Order in for的一个group啊行吧,好,那同样哈,我们还得再来一份啊。
07:23
呃,先把这个写完吧,先把这个写完啊,那就把这个order info topic name传进去,把这个order info,这个group传进去啊,好,这是我的order info的一个opposite。对吧,我就把它读到了啊,行,那还是再来一份啊叫all detail,它的这个topic的名字啊,OK,那我们也是把它复制过来。好,把这个info改成detail就行了啊。我们都是大写的啊,都是大写的,OK,再来一个啊,Order detail的一个啊DWD,然后呢,Order detail啊,然后呢,加上group啊行,那这个有了以后呢,下面还是我们的my opposite yous.read直接读了啊把这个all detail,呃,错了啊,Topic。
08:21
好,然后呢,All detail的go传过去啊好,这是all detail的opposite。行,读回来看到了吧,两份啊,都是两份,这个是order in for的啊。好,这个是detail,这么来写吧,啊,Info啊。看到了吧,行呃,那这个选好以后来接着往后第三步啊,就是从卡不卡中消费数据啊好,那你消费数据的话也是一样的啊,我们先去消费这个order in for啊,那就判断了啊,如果说你的order in for的这个opposite它是不等于空的,对吧,并且呢,Order in for opposite它的这个长度啊,就是应该是叫什么叫now empty对吧。
09:16
好,如果是这种情况,那我就可以直接MY卡不卡us夹叫get着卡不卡stream,把SS传过去,把你的orderf的这个topic name传过去啊,Order in for的go传过去,然后把opposite order in for的opposite传过去,好接收回来啊,这个是叫order in for的卡夫卡stream啊。对吧,我把这个提出来啊。好,放到这儿改一下啊。这是等于空的啊好,然后呢,下面就给它赋个值呗。好符合值啊行,那else的话就是。拉过来放到这儿,然后呢,这个东西就不要了。
10:03
对吧。来,下面就是我们的好几跳。OK,来,那再写吧,还是如果说你这个all detail the opposite它是不等于空的,并且呢,你的all detail的这个opposite,它的什么now empty,那就什么一样的啊,就是直接说麦卡us点带卡不卡stream传吧,All detailed名字all detailed。然后呢,Detail的这个。好,接收回来detail的一个,呃,卡不卡。上把这踢走啊。提到外面。改一下啊V好了,然后呢,把这个拿过来给它赋个值。
11:04
能看到吧,那else就是如果说你没有的情况下,那我就直接。过来,这个干掉不要了。OK吧?啊,这种代码我们写过,写过什么,写过三次,这应该是第四次了哈。对吧,应该第四次了吧,所以说呢,就不再跟你解释那么多了啊来接下来第四步提取outside,呃,这个t outside,我们先去提一下all in for,嗯。提一下啊transform好RDD,然后呢。Has instance of,然后呢,叫做has opposite range啊。OK,这个得改个名字啊,叫Allin for opposite ranges啊,把它提出来。
12:02
诶。好,等于空啊,行,然后再给它赋个值就行了啊。把支付到这最后呢,再把这个RDD返回回去啊,然后呢,这个接收回来,呃,取个啥名字呢,这个就叫做order in for the opposite的吧。好吧,这。呃,那这个提完以后呢,这个提这个提的是我们的还是一样的啊,这个提的是那个order in for啊都是两份哈,对吧,这都是两份啊,上面我们都说过对吧,两份两份两份。是不是来下面是我们的order detail啊,Order detail啊,来再写一下吧,好然transform r DD,然后呢,As of,然后呢,Has。好,这是找all detail的opposite range。
13:06
可以吧?提出来啊,放到这。好,这是VAR啊行,然后呢,再给它赋个值。OK,然后最后把RDD返还回去,好,那我们再接收回来。这是all detail的,叫off suchre啊,错了啊。OK啊,那终于呢,呃,就把这个该提的东西都提出来了啊,那提出来以后呢,接下来第五步啊,这个第五步呢,就是处理数据啊呃,处理数据的话,我们第一步就是还是转换这个结构啊,转换结构。对吧,那这个转结构的话,你看一下啊,现在我们拿到的还都是一个consumer record。是不是,那么接下来我们需要什么对它这个进行这个处理,那我们肯定还是只关心关心什么呀,我是不是只关心你的这个Y6啊,同学们说我把这个Y6提出来好,那你这个Y6提出来以后,你要转成什么结构呢?
14:05
你可以转成这个通用的,通用的就是这个,比如说map呀,对吧,或者说这个呃,再生object呀。都可以啊,或者你转成专用的,比如说你的order info,那我就单独声明一个order info的类,你的order detail,我就声明一个order detail的一个类。这也是可以的啊呃,那因为我们整个这个处理过程的话呢,还是相对比较复杂的啊,所以说呢,我们就给它转一个什么这个专用的结构,因为专用的结构我们用起来呢,会更加的方便一些啊,肯定比这个通用的结构呢,要什么更加的直观一点啊,因此我们在这个处理转化结构的时候呢,我们就这样啊呃,准备两个类啊同学们。来,我写一下,写到这个病里面啊,我们先用上一个SKY类,我们叫做order info,好,那我再来一个啊。再写一个啊,又上一个类啊,我们叫做order detail,看到了吧,行,那这个先来写这个all in啊呃,凹ino的话,我们就直接样例类好,那这里面有什么字段呢字段。
15:05
就是你当前这个拿过来这个流中啊,看一下啊,就是你当前这个拿过来这个流里面,这不是out的那个流吗?流里面啊,它里面所封装的那个数据,这个数据里面有什么东西,那我就要都把它定义出来,当然呢,你还希望你去补充什么维度,对吧?你都把它定义出来啊,OK,所以这个我就不再给你写了哈,直接去。复制一下就行了啊,这是我的order info啊,还是比较多的哈。好,拿过来啊。OK,看一下吧啊,这一回让我给他一个什么末耳值啊,这个其实看情况吧,你想给就给,不想给也行啊。好,呃,那这里面看一下吧。看一下啊呃,这些字段你可以自己去对比一下啊,这些字段我们在这个order info中呢,都是有的啊,来直接看表就行了啊order info。呃,看这个,看他的这个信息啊。
16:00
好,大家来看一下啊,这些字段它其实都是有的啊,呃,这还挺多的对吧?嗯。这个怎么看,这样这样来看,好吧,看一下吧,大家去对比一下啊,这个ID啊ID啊,还有什么这个。呃,什么order status对吧,然后什么UIDUID,然后什么total amount。呃,Total amount total在哪,在下面啊,在这啊,Total amount,对吧,就等等一些,我就不再挨个给你对比了啊,然后呢,很明显很明显就是它的东西呢,要比这个多。对吧,好,那这个也正常,因为什么?因为你的order info中啊,并不是这个每一个字段我都要啊,那我觉得这个地方的话,我们之前也说过,就是你数据你往这一层去处理的时候,那就看你将来这个分析的时候,你重点用什么,如果说有些字段我根本就用不上,那我是可以什么不把它放在我的这个olp中的。理解吧,啊,就是你把要的东西给他拿过来就行了啊好,那我们要的东西的话,就是这么些个,这就是我这个现在要要的就是从你的order info中要去出的一些什么字段,把它什么都搞出来。
17:04
能听懂吧,好,那除了这些之外的话呢。下面有两个这个。就是时间字段啊,一个是create data,一个是什么create hour,就是一个是创建的日期,还有一个什么创建的小时间啊,就一个小时能听得吧,小时啊小时OK,这个也是要要的啊。行呃,再一个就是我们的维度了啊,你看这个。是不是你的这个地区维度啊,你看这个。是不是你的用户维度啊。对吧,这两个的关联我们已经做过一次了啊,所以一会儿我们再去做的时候呢,其实可以完全复制过来去做啊,这个如果呃想写的话,再写一遍也行啊好吧,那整体就是我们要把你目前这个流中的数据啊,把它拿出来,拿出来以后呢,把这几个字段给他什么搞定。对吧,把这几个子弹给他搞定,搞定以后呢,呃,再把这个拿出来。再把这个单独处理一下,再把这几个单独处理一下,这不就OK了吗?听明白了吧,啊,这就可以了啊,行来吧,那我们就操作一把啊呃,那这个不对,还差一个啊,这个奥detail t还没做是吧?来DT拿过来。
18:08
Detail拿过来啊,Detail在这啊,它那边也是一样的啊,给上一个样例类,好放到这看一下吧,这个detail的字段它也是比较少的啊,Order detail。在这呃,它的字段的话,你看整体好像也比我们现在所看到的要多一点,对吧。比这个是不是也多一点啊。那就是一个意思啊,就是我们最后分析的时候呢,我们只考虑这么几个就可以了,对吧,像这个什么麦就什么URL。这些东西我们根本要都不带,不会要的。能看到吗?同学们,就有的东西呢,我是可以不要的啊,有的东西呢,我是得要的,OK吧,那我们就把要的给他保留下来啊,好,当然了,我刚刚也说过,就是你的all detail,我们目前看它里面好像没有什么维度需要你去,没有什么维度需要你去做关联,对不对啊,你看像我们的out INF的话,我是需要你关联一些维度的,但是我的out detail的话,目前是没有维度需要关联的,好,但是呢,不代表的说它就不关联。
19:05
那么你是可以让他去做关联的,比如说我想在我的all detail中呢,去看到什么呀,你的SKU更加详细的信息。虽然说啊,我在我的al detail中呢,已经耦合了。对吧,已经耦合了什么,耦合了这个SQ的名字了,就正常情况下,我的SQ就是我的每一个订单详情里面的,你买的是哪个商品啊,你买的是哪个商品,正常我只记住你的SQID就可以了。能明白我的意思吧,但是我们这个表的设计的话呢,已经是什么呀,把在它里面耦合了我们的SQ名字了。因为这个东西用的特别多。所以说你每次去做关联,每次去做关联的也不太合适,我们就这么直接把它耦合存储了,好,那当然对于我的SQ来讲的话,你看一下啊同学们。SQ info,除了你的什么呀?你看啊,除了你的SQ的ID对吧,SQ的名字之外,你看一下是不是还有很多别的信息啊,价格对吧,这个重量。
20:00
是不是还有什么所所属的类别。能不能看得懂啊,是否是一个销售的状态,然后还有什么,就是你的描述,我们还有别的信息,那如果说你在后续分析的时候呢,你需要用到这个SKU更加详细的信息,那你也需要把这些东西给我维度关联过来。明白吧,啊,当然目前我们是没有的啊,我们有这个SQ的名字就够用了啊,其实我们对于SQ来讲,我们重点还是看他的名字。听到了吧,重点还是看名字啊,OK,行,那既然它都是有的,那我们就不需要再去做这个关联啊,那我们只需要把这个定义好,定义好以后呢,一会把数据给它处理成这个格式就行了啊好,那接下来我们就把这个处理一下啊呃,那下面就什么转化结构了啊,结构怎么转呢。那叫什么,Order in for。Order in for一个什么叫做opposite stream啊,不要用错了啊,同学们表转结构我们用的是map对吧,那你map的时候呢,你给我传过来是一个record,好,那我给你返回什么东西啊,首先我要code.value提出来,这是一个就叫VALUE6吧,对吧?然后接下来呢,我们通过摘,呃,通过这个。
21:12
摘啊,因为它是一个摘数据啊点。先打个包啊。好,Pass object,然后我们转成对象,那么转对象的时候把value传过来,我要告诉你给我转成什么类型的啊,我们class of写一个order info,你给我转成这个order info类型的。All in啊,打个包。好。OK,那最后就拿到的是一个order info,那我再把这个order info给他返回回去,这就可以了。对吧,那么这样的话,它就可以什么呀,把你这个整个摘生数据中啊,跟我这个当前这个order info中所有相对应的这个字段,它就什么都给你封装进来了。明白吧,那你orderf photo没有了字段,但它里面有,那我不要,然后呢,我order INF中有的,但是你里面没有。那我就不给你封装,你就默认值就完事了。
22:03
因为那个维度的数据是需要我们自己去补充的啊,行,把它接收回来,好,那么这就是我们的order info的一个stream啊,行,那同样的道理啊,我们可以,呃,再来写一个啊,就是那个叫order detail。然后呢,Stream,然后呢,我们也做一个什么呀,做一个map操作好,你给我传过来的还是一个consumer record对吧,那我就什么record加value接收回来好,还是一个什么pass一下object啊好把这个value传进去,然后呢,再把这个class off。我们写一个叫order detail。诶凹的是吧,消息。来打个包啊。好,接收回来啊,这是我的order detail,然后再把这个order detail呢给它返还回去啊,行,接收回来,好,这是我们的order detail的一个string OK吧,同学们好,那写到这一步以后呢,我们就可以做一个打印了啊,打印上这个100行吧,先不着急往后写了啊,你先保证能够正常把数据给他拿过来,好来,接下来做测试,把它启动起来。
23:26
呃,那你下游要起的话,下游要起的话,那我的上游也得起,对吧?来贝斯启动起来。呃,因为我们是涉及到业务数据了,那我的。Maxwell也得起哈,把这个都退出去啊,Maxwell我也起一下啊。Maxwell好Maxwell然后呢刚刚卡,然后呢刚刚对啊行看一下啊。
24:00
呃,诶。这个已经起了是吧?DWF起了啊,贝斯DB没起吗?让一下啊。好,上游要起啊,上游起来,然后再起下游,下游都起,起来以后呢,我们就这个正常的去生成数据,然后看数据就行了啊好,这已经完事了,来完事以后我们就跑一把数据啊,那就直接DB.sh我们写个日期啊,杠零三杠二四对吧走。好,他开始跑数据,然后呢,我们就来到这儿看数据就完事了啊。诶。啊,有了啊,这个感觉好像有点小卡是吧?行,看一下吧,这个数据过来了啊,这是我order info的数据,然后这是我这个order detail的数据,来我们大概瞅一下啊,Order in for中你看一下啊,这个前面的字段应该都是有的啊,但是后面就没有了啊,因为后面这个字段是属于什么,是属于我们的维度数据啊,就属于我们后面的这些个字段。
25:15
对吧,属于这些个字段它肯定是没有的,这个是需要我们单独自己去处理的啊好,那对于我这个auto detail来讲的话,它里面应该都是有的。能看到吧,因为它是不需要关联维度的啊,就说白了,目前你的这个all detail中所有的这个字段,我在数据中已经有了,我就什么直接把它给你封装到你那个对象中了。OK吧,行,那这个我们就呃写完了啊,就是我能正常什么消费到数据了,然后接下来我们就再做后续的处理啊,好听一下。
我来说两句