00:00
怎呀,对吧?注意啊,咱们整个的业务,对吧,大家一定要跟着我思路是吧?呃,那么咱们现在呢,把这个东西对吧,这个偏移量啊已经获取到了,然后呢,从我这里读的数据了,然后把到现在呢,这个偏量位置也得到了,对吧?那么这块的业务呢,其实咱不用变,对吧?还是做结构的转换,然后呢,这里还是做咱们去重,对吧,然后接下来这里还是往ES里边来保存,对吧?那么等这些都做完了之后,大家想一想,我现在是不是相当于我的业务处理完了呀,我已经把数据保存到我的ES里边去了,说明我现在这个数据呢,已经落盘了,已经处理完业务了,那么处理完业务之后,那么按照咱们现在它的一个精准性,一次性消费,它的一个保证,咱们现在相当于说这里我已经执行过了呀,咱们现在是不是可以修改偏移量了呀,对吧,那么如果修改偏移量的话,那大家想一想,咱们现在在什么位置去改呢?电源加在哪呢。这个括号要加加好对吧,那么你一定要把这个层次给弄明白,大家想一想,我们现在呢,在这里执行这是什么?For each r DD,对不对,然后接下来,那么咱们现在的这里是不是以分区为单位来对咱们的数据来进行处理啊,对吧?那你想一想,那这个东西加在哪呢?是加在分区这儿呢?还是加在咱们上面这块。
01:23
加在那是不是应该加在上面了,对吧?大家想一想,我现在在改的时候,我是改的分区吗?不是我改的是不是当前这一个批次,然后你从卡卡里边读到这个数据的便移量是多少吧,对吧?那你对批次型操作,那你想想这个RDD是不是才代表咱们当前这个批次呀,对吧?所以说呢,那么这块呢,是我们当前呢,对整个的批次来做一个们提交操作,对吧?这个的参数这个呢是我们结束的位置,所以说呢,那你应该在这里对来干什么来去提交我们这个偏移量啊,提交偏量到哪呢?到我们这个ready中对吧?那么如果提交的话,大家想一想,是不是有一个咱们专门管理偏量的工具类呀,Outside manage u,然后接下来咱们有get outside方法,现在除了get之外,再来一个save,对,用它呢去保存我的偏移量,那么如果保存偏移量的话,大家想一想我需要提供的东西,你是不是得告诉我咱们当前是哪一个主题?
02:24
你是不得告诉我咱们的消费者组的是什么,你是不是得告诉我咱们现在的这个outside ranges说这个东西你得告诉我对吧,其实这个东西呢,注意主题呢,其实你可以不传。对,因为咱们现在这里,我我我是不是从这里也可以获取到了,对吧,这里你看咱们刚才在我这个数组里边,对吧,它里面放的是outside range,它这里面是不是也有topic对吧?也可以啊传过来也可以,我这直接传过来了,导这直接传过来了,然后接下来那么咱们现在呢,那把这些东西呢,给我传过去。啊,我们都会传过去对吧,那我现在呢,要提交我们现在的这个便宜量,然后到我们这个red里边来,那目前没有这个方法呀,对吧,目前没有这个方法,没有这个方法咱就创建一下呗,咱在这里我们呢去创建一个这个方法,对吧?那么这个方法呢,它所完成的功能对吧?将偏移量信息保存到我们这个rap中,那么具体这个保存,那么怎么来实现对吧?具体这个保存怎么来实现对吧?那么大家呢,先有一个思路对吧?那么没准呢,一开始思路呢,可能连贯不上,但是你先按照你思路啊,把这代码敲出来,敲着敲着没准这个思路就出来了。
03:36
啊,敲着敲着没准思路就出来了,对吧?比如说我现在呢,要想去做咱们那个保存的话,对吧,那么我现在呢,本身在这里,你要想保存的red里边去,那首先你是来先获取咱们这个连接对吧?哎,首先第一个啊,在这里get我们现在呢这个连接,那应该是我们从我red来获取吧,那就my ready尤点get je class.vr对吧,那么这个呢,是我们je客户端,然后接下来je客户端是不是得去关闭一下呀,Close,然后呢,那你要去执行,执行什么呢?咱们现在是不是要往里面放啊对吧,那你来到这里啊同学们对吧,你要往里放的话,同学们你现在是什么,你现在是我们这个对吧,哈希类型了,那么怎么样往里放呢?对吧,那咱们现在把这东西都提一起,对吧,把该起的东西啊给提起来。
04:33
把这该起东西起一下。连着了是吧,然后提起来,那我现在把EQ都把它提起来吧,对吧,那肯定CK来提起来当了。啊,卡不卡,再起起来。
05:04
那么起完之后呢,咱们这个ES是不是再提起来的,ES再提起来,那么咱们这个log是不是也得提起来呀,对吧,这个log你得处理我们这个质嘛,把这个log那也得提起来,除了咱们这些之外,这个东西red是不是得起呀,对吧?来到我们的o BT model,然后接下来red,我们的一个server啊,Ready,这应该是三个对吧,然后接着呢,在咱们这里边,我们去执行我们这个RA然server,然后接下来RA Co,对吧?然后呢,我去登录一个ready客户端,对吧?然后接下来咱们这个case。啊先对吧,那目前呢,咱们这里来看,这里是不是有我的便宜量啊,对吧,但是这个东西呢,是我昨天随便瞎写的,对吧,那么你现在要往这里面放数据啊,然后你放数据,怎么放数据呢。
06:00
咱们当你是哈西,然后接下来往里放出去,对吧?HS,然后呢,你K对是什么,然后呢,Fail的什么对吧?在这种放法,同学们你是不是只一次能放一个呀,对吧?你这一次只能放一个,但我现在一次想放多个,因为你想咱们现在这里拿到的东西,你看对吧?在我们这里我拿的东西是什么呀,是不是一个数组啊对吧,也就是说当前这个主题对吧?然后呢,它有多个分区,零二分区,一号分区,二分区,三号分区,那么零号分区的偏量是多少,然后一号分区偏量多少,二分区偏量多少,三号分区偏量是多少,那么这个呢,对吧,我现在想一次都放进去。啊都放去,那么这个时候你不应该用我们这个set,那这里是不是有一个呃操作多个的方法,有set吗?对,来看ATMS哦HMS那进来K,对,然后呢,加上咱们这里是不是多个我们现在这个这个名这个预值,对啊对吧?那么如果说这块啊可以用我们这个HMS的话,那接下来咱们呢,来那就到我这里边来呗,对吧,我现在呢,要想去操作,那你需要调什么,调它的一个HMS这个方法,那么HMS这个方法呢,第一个是不是需要一个K呀,对吧?那么目前这K没有对吧?那咱们现在呢,先拼接我们的这个RA中的来中来操作我们的偏移量它等于K啊偏移量啊它等于K,那么这个K呢,肯定和上面这个东西是得一样了,对吧?所以说呢,这块咱们直接CTRLC把它拿过来啊,定义一个我的outside k,然后这里边的outside将了主题。
07:39
加上咱们这个消费者组,然后把这个呢给答过来,那么除了它之外,那么咱们现在再往下放,其实是不是就是我们现在它的一个每一个分区的偏移量是什么呀?那你看一看人家这个方法,它把这个东西封装成什么呢?说封装成一个了,我们现在它的这个map部集合呀,对吧,而且这个map部集合是谁呢?是不是咱们Java的一个map集合呀,所以说接下来那么咱们在这里干什么呢?对吧,我现在要定义我们一个Java它的一个mava集合,那么用于干什么呢?用于存放咱们这个分区和偏移量啊,每个咱们这个分区啊,以及呢,对应的偏移量吧,对应的偏移量啊,我要这个东西呢,要把它存一下啊,要把它存一下,那么如果说你现在这里加上集合的话,那就来呗,啊,那我在这里new一个对吧,然后呢,U点哈西对吧,麦。
08:35
啊,那么这里放的什么呢?这个是分区,这个是偏宜量,对这偏量,然后VR咱们现在比如说叫做set map对吧,S outside map对吧,这时候关于我偏量的一个集合,短关偏量集合,然后接下来咱们现在呢,可以把outside map呀,给它拿过来啊outside map给拿过来对吧?那么咱们现在呢,在这里啊同学们对吧?那如果说你这里直接往这里拿的话,这里边好像是少东西吧,对吧,我看一看在到们这里边把这个文拿过来,点我们这个VR对吧,然后接下来注意看啊同学们,咱们现在呢,我想我的想法是这样的对吧,然后呢,这个是opposite outside map是吧,Opposite outside map,但是呢,人家在这个要求的时候,你看这个集合什么类型啊,里边是什么string string是吧,你看string string,所以说呢,咱们现在这里一定浪不太合适对吧,来把这改了对吧,把这改了对吧,然后接下来outside map再往这地方放啊,这可以吧。
09:35
但是你现在是空的呀,对,你是空的,你是不是得去什么呀,从我们这个off range里读取数据呀,你想想我现在的偏量信息是不是在这里放着呢,对吧?哎,它是一个数组,这数组里面记录的哪一个主题,哪一个分区偏量从什么开始到什么结束,所以说呢,那我现在应该什么对吧,我应该对于我们当前的这个来进行便利啊,来对当前当这个它来进行一个我们这个便利。
10:05
那便利,那便利完之后呢,来将我们这个数据来封装到哪?封装到我们这个outside map中对吧?封装outside map里边对吧,那就来呗,Outside,咱们现在这个ranges点我们的一循环对吧?那么每次遍利呢,它得到的其实一个outside range对吧,这样一个对象对吧,Outside range对象,那么你通过outside range都可以拿到什么东西呢?对吧?那么咱们看一看在这里outside range点同学们在咱们这里我可以拿到什么partition是不是可以拿到这个partition拿到了吗?那首先第一个这个呢,是我们的partition ID啊,这是分区号呗part ID,然后接下来outside range点咱们现在这个from outside这书可以拿到,从什么地方开始对吧?然后还可以拿到什么呢?Offet read.until对吧,到哪结束对偏音量嘛,对吧,那么咱们V。
11:06
啊,这个呢是我们的N对吧,然后接下来我们的outside,那么你拿这个东西之后,对吧,这不行啊,你首先把这东西往里放啊,对吧,其实我们现在呢,同学们这里边我要放的话,我都放什么?第一个咱们现在要put,这里面其实放两个内容,一个呢是我们这个分区吧,那么另外一个是不是当前分区对应的一个我们这个结束便移量,所以说这里面其实我拿两个东西就可以了,一个是它对吧,那么另外一个呢,是我们这个按until outside,只不过呢,人际要求是不是咱们都是字串的,所以说咱们在这里转换一下to string,然后这块呢,也去转换一下to string,对吧,那么咱们现在呢,在拿到之后呢,我在这里对吧,放完了我做一个输出对吧,做一个打印输出啊,上面呢是读取分区,对,那么现在呢是保存啊,咱们这个分区对吧,那么保存咱们这个分区是什么呢?这里盘的是ID对吧,给它拿过来。
12:06
然后接下来,那么咱们现在这里再加上一个对吧,我们现在呢,是从哪开始对吧?从哪个偏移量开始,然后接下来那么到哪个偏移量啊结束对吧,这个呢,是我们这个到这对吧,结束对吧,Until outside对吧,咱们在这呢,做一个测试啊做一个测试好了,那这样的话,咱们现在呢,执行完毕之后,同学们把这个数据呢,我封装了一个map集合,把map集合里有了,然后接下来那么咱们呢,去调用我们这个HMS的方法,然后把咱们这个零数据呢,给做一个保存,保存到我们这个啊软上去。好了,那这样的话呢,咱们在这儿短关于啊去保存我们平量的操作,那完事了,那保存片就完事。整体测试呗,对吧,测试的这个才有点意思,对吧,光这个代码这个写的没啥意思,一起来测一下同学们。
13:08
呃,那么如果说呢,我现在想测试的话,那在咱们这里边对吧?呃,首先那干什么的flash的flash,然后接下来那我现在呢,来到我们这里啊,来到我们现在这一个T班上啊哈豆202啊断号5601对吧,去连接一下我的K班。然后呢,那目前在咱们这个K班的里边去找谁呢?对吧,J茂对相关的昨天对吧,我产生了一个数据对吧,昨天啊,这是我昨天产生的数据啊,昨天产生数据那一共呢,是50条,对,一共是50条对吧?那么接下来,那么咱们现在呢,那在这里我说要生成今典数据了呀,对吧?那么如果说成经典数据的话,我再克隆一个我们session c do VT,然后呢,Model r tap log,对吧,然后接下来我们去修改一下我的prop文件,对吧?那么这里呢,我们改成24号段。
14:22
1024啊,今天过节多圈点代码是吧,那么咱们现在呢,把这个东西改完之后呢,那我现在开始运行了,同学们,那么运行的话,你是不是得先把这个东西给运行起来,对吧?这里我现在该清的这个已经清了,对吧?然后接下来呢,那我先把它呢给运行起来,那么如果它运行的话,大家想一想大概怎么走的,会从ready来读取便宜量嘛,Ready啥也没有对不对,那所以说咱们现在的你要输出,你看到咱们现在这个读取分析偏量这个操作你应该看不到。
15:00
等于ready没有啊对吧,所以说那我说那你说我会直接保存这个吗。肯定会呀,如果说我现在呢,第一次在执行的时候,大家想一想,他肯定这样,首先呢,从这里获取偏移量,但是我发现没有,没有怎么办?没有是不是从咱们已经读过的最新的位置开始读啊,对吧,那最新的话,那我当前这个主题这个分区从哪块开始读的,到哪结束的,我是不是也会记录到我们现在它的一个我们的outside range里面来呀,对吧?那么整个咱们程序在提交的时候,它是不是也要往咱们现在这个ready来保存呀,对吧,来看一看啊同学们,那我现在呢,去运行我们这程序对吧,在这里把这程序呢运行一下。好了,大家看咱们现在呢,这里是不是有我们现在这个保存偏移量啊对吧,只不过这个时候呢,你看每个都不变的嘛,对,比如你看二号分区637对吧,这个也是637对吧,每每每五秒钟都要保存一次对吧,每个采集周期呢,都要去保存一次,然后接下来那我现在呢,那我可以这样了,我呢去执行我们这张了杠这然后接下来这一帽对吧,来执行我现在模拟成数据对吧,来看一看啊,咱们现在这里这呢size对吧,咱们现在来看是不是开始向ES有插数据了呀,对吧,开始向E查数据了,而且大家注意看啊,咱在保存的时候,原来的零号603,现在零号对吧,现在零号620了嘛,是不是咱们当年偏量是不是也在发生变化呀,对吧,原来是吧603,你看这这里边对吧,这个零号原来620,现在630了吗?对吧,说老师那你这里这差三条怎么变630了,注意啊,它现在这个产生的这个偏量和你这个ES插入没关系。
16:54
对吧,ES插入是不是已经我从100条数据里面可能过滤出两条给插入进去了,但是呢,你实际从咱们卡卡里边是不是可能读到100条数据,对吧?因为你模拟生成数据可能生成很多条,对吧,都放在卡卡里面去了,我会把这些数据都能卡卡补出来,但是呢,我会这些数据作为过滤,对吧,然后呢,只会把咱们这不重复的给拿到好了,那么我现在呢,在这里基本上不怎变了啊,那我要干什么呢?
17:19
那我现在呢在这里,那我看对吧,这个还子生成了,我就不让它生成了CTRLC结束对吧,那么它解说首先呢,咱们要来到这里来看一看我们的K星。都有啥呀啊几个两个是吧,那一个是咱们日活的,这个日活什么类型的呢?DU20201024,这个什么类型的呢?对吧,那我想想查看里面东西怎么查看。S members对吧?然后呢,咱们把当前我现在这里面东西呢给拿到对吧,来看一看咱们现在呢这个日活对吧?诶现在你看这里边这个设备是不是一共是50个呀,对吧?然后接下来那么这块呢,大家看啊,我当前这个呢是我主题对吧?这个呢计某什么东西。
18:08
我我没写记吗?对吧,看看是吧,在这里,那我就得改呗,短其实这个东西呢,你这么叫他也没问题是吧?但是呢,这就感觉不太好,对吧,来到这里对吧,这眼神对吧,那咱们现在呢,是我们这个地贸对吧,地貌对吧,然后接下来呢,那么咱们啊这个完事了,对这完事了对吧?哎,那我现在呢,看一看它这里边东西啊,其实这里边呢,那主要放在什么数偏音量信息啊对吧,那如果放偏量信息的话,那这个怎么看呢?是不是有一个什么H,我这get套是吧?对h get套,然后接下来把咱当前的K呢给拿过来啊,把当前咱前的K给拿过来,对,咱可以拿过来,然后拿来之后咱现在再来看一看,到在这里执行一下啊,那这里一共呢,是我们四个分区0123,那每一个分区对吧,它的偏移量是什么对吧,是不是记录下来了对吧?哎,每一个分区的偏移量是什么,咱们这个呢,就。
19:08
就下来了对吧?呃,然后接下来除了看这些之外呢,那么再看哪呢?是不是还得再看我们当前这个一来算式里边的数据过没过来呀,对吧?那么咱们现在呢,要看elect search对吧?那这里应该有个j mo1024,注意啊,这个1024当前我现在查的谁,是不是就查的我这个索引呀,对吧?哎,那么这里一共多少条呢?50条,那我们说我现在想查询十月份一共对吧,有多少个设备访问了对吧?这十十月份啊,一共啊有多少记录访问了吧?对吧?那么这个时候咱们是不是可以这样get呢?谁呢,我们可以通过别名查吧,对吧?那么这里有什么g Mo,然后接下来是不是可以直接上这个块一号,然后接下来做一个search,那么这个时候会把它们两个一块查出来,对,把这两一块查出来就是100个对吧?把这两件数一块查出来对吧?这也是我们现在别名的一个作用吧,啊,别名一个作用相当于把它们做成一个分组对吧?我现在想把这个数据都查出来,也可以通过别名对吧?来做这个事啊,来做这个事好了。
20:08
那这样的话呢,关于我们现在这个偏移量对吧,处理这块呢,那就完事了啊,关于咱们这个偏移量这个处理这就完事了,对吧,那么这里呢,我先停一下吧,对吧。
我来说两句