054_尚硅谷_实时电商项目_编写SparkStreaming程序消费kafka数据

在这呢，我先去创建一个我们这个IP啊，这个呢叫我们这个APP啊，那么第一个APP呢，咱们这个名字呀，就叫DAUAPP就可以了，对吧？那么日活啊，它本身呢叫daily active user啊，就咱们日活用户，所以说呢，在这里我呢先去创建一个scner这样一个程序啊，然后接下来在这里啊，这个呢是我们这个日活的一个业务啊，日活的业务，呃，那么如果说要想完成到日活的话呀，那在这里我呢，是不是得去从卡卡里把这数据给读到了，对吧？那我现在要干什么呢？我要从我们这个卡夫卡中，对吧？从我们这卡夫卡中来读取我们数据，谁去读呢？那是不是应该咱们SPA streaming啊，对吧？通过我们这个SPA streaming。来从那看看的数据，那么如果是SPA的话，那么这个时候呢，咱们首先呢，是不是得先创建这一套东西啊，有一个我们这个Spark Co啊，又一个Spark Co，然后接下来那么咱们呢，设置一下master啊。

然后接下来设置我们这个APP内啊，APP内这个呢，就叫它就行啊，然后接下来点对压这个这个。然后接下来咱们呢，要创建我们这个string contest啊教咱们这个string contest，又一个我们这个streaming。康带对象传两个参数进去，一个呢是我们配置文件啊配置对象，那么另外一个呢，是我们这个采集周期啊，那么这个采集周期呢，咱们指定比如说呢五秒，然后接下来这个呢，是我们这个SSSC的SSSC，然后呢，你要想从我那卡法读出去的话，工具类咱们是不是有了同学们对吧？那所以说呢，我直接去调用一个叫做MY卡不卡对吧？MY卡YouTube其中呢，有一个方法叫做get卡卡stream，这是一个重写的方法啊，这一这是一个重载的方法，对吧？那么咱们现在呢，传这么几个参数进去，首先第一个topic你得我传过来，那么第二个S你得给我，那么第三个group ID也是得给传过来，对吧？那么关于咱们现在这个偏移量呢，这个咱们现在没有对吧？那我现在先传咱们这三个吧，对吧，那目前topic和group ID没有那VR。

Topic对吧？那么咱们呢，得去定一下string类型对吧？Topic是什么？然后接下来v are group ID对吧？那么这个呢，也是一个我字串对吧？那么它是什么？咱来定一下，那我当时在咱们这个里边我的卡卡里面对吧？它的一个我们的topic名称是什么是吧？忘记了，那咱们现在把卡卡提起来是吧？那CK，然后呢，Start启动我们组keepper啊启动组keep，然后呢，启动我们这个这个卡普卡啊，启动咱们这个卡夫卡。

然后看一看咱们这个启动情况啊，看一看启动情况，然后呢，都启动成功之后，来到我们的OPT model卡卡下边，然后接下来我现在呢，想查看一下我当前卡夫卡里边都有对吧，这个哪一个哪些主题对吧？那么这个命令应该是什么？来勾南星同桌啊，不能再叫你了，是吧？找同桌就行。

赶快下班，卡不卡？我想查看当前咱们现在呢，这卡出来里面都有哪些主题，你就不能随便改啊啊，然后接下来杠杠keep，然后接下来继续，然后。多少202，然后呢。端口号9092。杠，那现在执行一下。

等待呗，对吧，等待，那么等待到了一个时间之后呢，这个应该是报错了对吧，就说老师我现在用组keep行不行，同学们可以是吧啊，但是呢，注意啊，同学们，这个如果有组keep的话，你的端口号呢，也得改一改，等端号就改一改，这个咱不用等，等时间长了，到时肯定连不上咱组keepper对吧？但是大家可以看一看咱当前的命令啊，你可以看一看当前的命令对吧，就是我现在呢，为什么不用咱主keep呢？因为它现在目前啊，已经标记为咱们这个过时的了，对吧，就在咱们现在呢，在标记过时，在咱们这个卡夫卡topic命令里边呢，还能用对吧？但如果说我现在在咱们那个什么卡循环命令里边，这个东西它已经用不了，就直接把它干掉了，早就在早期版本里边，你不管是我们现在讲什么啊，Topic，或者说是什么consumer啊，这个啊，Consumer它都可以用，咱们走keepper啊，但是。

后来们现在这改了，这应该是我们对吧，那注意啊，如果说咱们现在这个命令呢，还对吧，没记下来的话，对吧，那大家下去呢，这两天抓紧记，如果下次我不小心我再去交育了的话，如果这个命令还有问题，对吧，那有点不合适了是不是对吧？好那那现在呢，在这啊，我现在把我们这个呢打过来，我是不是只需要去拿咱们这个启动的，我们这个这个topic启动制topic就可以啊，所以说呢，那来到这里，咱们呢，把这个拿过来，那另外一个呢，我现在有一个主题对吧，这个消费者组，这个消费者组呢，咱们把它拿过来对吧，拿来之后呢，改一改咱们这个加一个后缀是吧，623对吧，那这样的话，咱们现在这个东西是不是有了呀，那么有了之后呢，点2V2一下，它的返回值呢，应该是什么？应该是。

第一个我们现在大家看一看，是不是应该是一个我们现在第一个我们的第三化了，对吧，那比如说叫卡然那这个东后，那么们定去什么去开始去采集了吧，那就start的，那接下来咱不断的关对吧，那么这块呢，来对吧，来调一下，然后接下来那么我现在呢，要想去这个输出，那么输出的话，注意同学们你从咱们这里边拿的呢，其实我现在这里是不是一个我们这个consumer card的对象，所以说呢，咱们其实要不了把这么多，把只要其中的value模式不就可以了，所以说呢，在咱们这里边我map一下把谁呢，把我们这value取出来啊，把value取出来，那么这个value呢，那里面放的内容是不是其实就是我们的Jason字符串呀。

因为你现在往你这个主题里放的是什么，是不是你生成的一个我们的启动日志，是不是一个接的字串了，对吧？那我现在要把这个东西拿出来对吧？所以说呢，我只取它的一个Y流部分啊，我只取它的Y6部分对吧，那么取完之后呢，那么咱们把这个东西啊，给它输出出来对吧，在这儿是吧，那我现在呢，这个做一个输出对吧？把这个提示也给你拿出来吧，这个杰森是吧。它本身也是个stream啊，那这个stream呢，那应该是我们的Jason啊。然后接下来咱们现在呢，请调J它的一个print的方法来做一个打印输出，对吧，那到这为止大家看一看这块代码啊，能不能理解，感觉每一个东西代码都不太多，说是啊，然后接下来咱们现在啊，代码不难，难的是同学们我要看效果。

我现在呢，我要先干什么效果呢？对吧？模拟器生成数据，然后呢，你得有专门的架包对我的日志的来进行处理，处理完之后呢，往卡卡写我这个程序从卡卡它的一个主题中来把这数据给读出来，对吧？那么咱们现在要想做这个测试的话，同学们，那我在这里得得得干什么呀？对吧？来你看一看SQDPS对吧，那怎么操作。首先先先先把这个东西咱们先提起来，对吧，先把这东西记起来，我说不得等着去卡卡学完数据啊，但是现在卡卡里有数据吗？对吧，有点历史的，咱们是二的，那应该消费不到。对吧，那你要想去消费的话，那么咱们现在是不是得往这里来生产数据啊，那生产数据的话，那这个时候怎么来生产，提哪些东西N这个在起对不对？然后呢，咱们现在每一个我们处理日志的程序是在起啊，咱们是不是有一个脚本叫什么log是吧？然后呢，咱们把它给启动一下，那么启动完毕之后呢，咱们现在再来看对吧，那这里呢，那应该是对吧，把当前啊每一个节点上是不是都运行了一个我们这个处理日志的这样一个下包啊对，然后接下来咱们是不是得去生成我们这数据啊，来到o model来目录下边RT对APP咱们这个log下边，那需要去改一些东西，改什么呢？Public practice对吧？改我们现在业务它的一个日期对吧？今天呢，是我们这21号对吧，那我现在生成我们这个21号的数据，对吧，然后接下来咱们是不是可以直接执行了呀，当然后呢杠这执行我们现在进模模拟实施数据。

对吧，那模拟数据的话，这个时候呢，咱们现在来运行看效果啊同学们。三个袋子是不是成了呀，那么你要看谁呢？你是不是得回到这里来看呀，对吧，咱们现在这个数据对吧，是不是有了对吧，咱们现在这个数据是不是有了呀，对吧，有了之后，但感觉点是不是没给他输完嘛，对吧，没给他输完，那么这如果没给说输完的话呢，那咱们现在把这个东西先停一停，对吧，先把这个停一停，把这个呢也停一停，对吧，那么这个打印输出的时候呀，那么这里呢，应该有一个参数的，我记得对吧，那应该有一个参数对吧，来看看在这里对吧？呃。

Print是吧，然后看一看咱们当前再去执行的时候，对吧，他应该有一个，如果说咱们那个过多的话，对吧，那么咱们现在呢十啊。呃，我想想这个参数熟了呢。去掉这里一个打印啊，然后打印呢，这里面有一个参数，这个参数就我们一直想不起来了。

啊，应该是个布尔的参数对吧，但不应该在这里面传对吧？呃，在这儿看一看啊，它这打印。把这个呢，数据打印大一点对吧，打大一点行吧，咱现在这里我一会想一想这个参数是什么，我又想想参数什么，它其实是什么呢？就是说他在打印的时候呀，他会去做一个我们这个判断啊，他在打印的时候呢，其实会做一个判断，然后判断一下咱们当前呢，他这里边的一个内容，但其实在这。短缺在这对，呃，应该有一个参数设置，我想一想啊。然后咱们这边呢，还是让他去生成我们的数据，对吧，还是让他去生成数据啊。

那么这边呢，是我们这组size是吧？呃，那每次呢，如果说我现在要是联网的话，这块呢，它其实是这个这个连时间比较慢，到时候我可以把这个电话关掉，当概现在如果说你现在这个东西拿的对，如果说要是每一次啊，你现在处理的这个数字的个数对吧？然后和你现在采集的个数，如果要是比它小的啊，那这个时候数据不出现那个点点零对吧，我记得有参数设置的啊，到时候我再看一看，到时候再看看，嗯，然后接下来，那么咱们现在呢，在咱们这里对吧？在咱们这个输出的时候对吧？那我其实可以采集到我现在这些内容对吧？给以采集的内容，那么采集完之后呢，你可以把这个东西啊CTRLV对往这放一放对吧，那我们现在呢，再往下去做点东西啊，再往下呢，咱们再做点东西，我把这个呢，暂时先停掉啊，暂时先停掉对吧，那么你现在呢，再去做东西做什么呢？就你不能直接上来就把这个输出啊，对吧，输出完这这应该干什么呢？简单。

再去做一些处理啊吧，简单做些处理对吧，怎么处理呢？就是我现在拿到的是一个我们这个阶层对象啊，拿着接对象，然后接下来那我从咱们接子项里边呢，我可以拿到一个时间戳，但是我现在最终要统计的时候，我不是说我要统计日活嘛，日活你是不是要给我咱们一个日期啊和小时啊，对吧，所以说呢，那这块呢，我希望什么呀，我希望你把这个日期和小时给我。那我希望你把这17个小时给我。那么如果把这七个小时给我的话，那这块咱们应该怎么来做，就是本身我拿着进制项，基本这里面是没有这个东西的啊，我其实没有这个东西的。来是吧，那么咱们现在呢，把这些东西对吧，为了让大家看得清楚，对吧，还是来到我们这个Jason这个里边，然后呢，我做一个格式化，做一格式化，那目前咱们拿的数据啊是这样的对吧？其中呢有一个属性，对这个属性叫什么呢？这个属性叫时间戳对吧？那我现在呢，我需要对我们这个实验戳呢做一个处理对吧？来在这里首先呢，我要从对吧我们卡夫卡中对吧卡夫卡中来读取啊我们的精神，那么咱们读取的话，其实这个东西已经读到了啊读到之后呢，那么咱们现在啊这里我需要对它呢，做一个我们的映射对吧，怎么映射呢？就是卡不卡map对吧？那么这个map呢，他拿到的其实是什么，其实是不是一个我们这个对吧，这个T格式的一个字串，对吧，拿这个JA字串，但是呢，那我拿这个字串之后呢，那这应该不是字串了，就在这里，我应该Jason格式一个stream，然后stream里放的什么呢？放的是字串。

对吧，然后接下来呢，我需要对它呢做一些处理，那么如果处理的话，那咱们是不是应该想办法把它转成我的接对项，对吧，想办把它转到接对象。

等。那如果要做这个事的话，对吧，我现在最终呢，我不想是字串的，我想把它转成竞争对象对吧？那这个东西咱们在这应该是对吧，把这个拿过来注意啊，还是我们现在卡不卡team，这个其实是不是这个时候还是一个customer card当封装的呀，然后接下来呢，做一个卖的，做一个mind，然后接下来，那我现在首先第一件事什么呢？那应该是我们现在找大参数呗，这个是我一条记录对吧，Record是吧？那么咱们现在要干什么呢？通过咱们这record呀，它点我们的value，那么拿到一个值，这个呢，其实是不是就是我们现在这个JS格式的一个字符串，对吧，这个是JS字符串，然后接下来对吧？那么咱们现在呢，要将我们这个J森格式的字符串转换为G森对象，怎么转换？

怎么样把这个字符串转换成对象，是不是咱们这个fast杰son，这可以帮他做这个事了，对吧？那咱们现在呢，把fast杰森对吧这个拿过来，其中有一个我们这个的方法叫什么呢？叫做pass object，你可以把咱们Jason搜串给我拿过来，那么他拿到的是返回的是一个Jason object对Jason object对吧，拿到一个接对象，那么你拿了接对项之后呢，同学们，那其实你相当于拿到的是什么，是不是拿到的是整个咱们这个直间对象对吧？那你是不是可以从它直对象里边来获取我现在这个时间出呀，对吧？所以说呢，那接下来我要从咱们这个J森对象中来获取我们这个时间出啊，来获取我的时间出，怎么获取呢？J森M这个点，那么这里咱们要想获取时间戳的话，其实是不是可以通过咱们的get，这时间戳什么类型？

就这个东西，它什么类型的date这个都不是date long类型啊是吧？所以说呢，在咱们这里我可以直接调咱们那个Jason object.get lock这些方法，谁呢，把TS拿过去，然点VR拿到的是我们现在TS对吧？拿到的是我们现在这个时间戳啊，拿到时间戳，那么拿到时间戳之后呢，那你是不是想把它转换成我们这个对这个日期以及小时啊，对吧？所以说呢，那咱们需要将我们这个时间戳转换为我们这个日期和小时，那怎么转换？是不是得用一些咱们第三方的工具类啊，这个一般在做时间转换的时候，这个第三方工具类你们用的是什么？Single permit是吧，那你来呗，New一个我们这个simple date permit对吧？因为simple for，那么这个simple data for里边好像需要传一个咱们这个parent进去，对不对，那应该是年四位的月对吧，然后呢，这个日的年月日，对，然后接下来再往后这里是不是小时啊，对吧，那么咱们现在呢，可以达到一个这样的一个。

日期短这样的一个工具类，那我希望呢，到时候把这个东西啊转换成这样的一个格式，怎么转换呢？在咱们simple data form里边，它呢，其中有一个方法的，其中呢有一个方法，那么这个方法呢是format，这个format的作用，你可以给我传一个日期过来，然后呢，我给你返回一个日期的对吧，对应的这种格式的字串，所以说我现在我是不是需要一个日期呀，对吧，那这个日期怎么给呢？你可以new一个date嘛，对吧，只不过默认情况下咱们在new date的时候，对吧，我们用的是不是YouTube date呀，对吧？那么你现在这样的话，它获取是谁当前日期嘛，对吧，咱们如果想获取你生成的一个日志日期的日期的话，你是不是应该把咱们TS作为参数给传过去啊，对吧？那这样的话呢，它创建一个您产生日志的短这个日期这样的一个对象，然后接下来通过咱们这个工具类把对象呢，给转换成了我们。

现在它的一个周串对吧，那接下来点VR对吧？那么这个呢，应该是date塔对吧，那我把这个也拿过来了，给它加了一个日形式，我加了一个类型点VR对吧，那么这个呢，应该是我们那个date s tr对吧？那么如果你现在拿到咱们这个日期了，这个日期呢，长这样，2020杠，然后呢，幺零杠二幺，然后接下来这个是14点对吧，是不是拿到的应该是这样的一个一个格式的一个日期啊这样的一个字符串，那么你现在要干什么呢？你是不是要把它分开呀，对吧，你要把咱们这个天还有谁呢？还有我们现在把它的一个我们小时给分开，所以说你要想分开它的话，那这个得怎么分开。

那就split一下呗，对吧，Date s t点我们的一个split用什么呢？用空格做一个分割对吧，那么这个时候它拿到的应该是一个date s tr啊A是不是应该是一个数组啊对吧，然后接下来咱们现在呢，定义一个对吧date这个date dt啊那么咱们现在呢，通过我们这个数组对吧，可以拿到它的第一个元素对吧，数组第一元素零呗，对吧，然后接下来那么这个数组零其实是不是就咱们这个天呀，对吧，那我先定义变量对吧，那么这个呢，就是我这个天对吧，然后接来我再定义变量，这个呢，HR对吧，这个等于什么呢？对，HR对吧，然后它的一对吧，那么你先把这个东西拿到之后呢，我的目的是不是想给咱们这个接似对象给它加过去啊，所以说呢，这个时候咱们可以调用J森of这的点to的方法，对吧，来给它呢添加两个属性，一个呢，是我们这个DT对吧，这是我们这个。

It对吧，哪天对吧，那么另外一个呢，咱们就是我们这个小时对吧，另外一个呢，就是我们这个小时对吧，把这个HR直接拿过去对吧，然后最后把咱们d object作为我们现在这个当前map的这一个返回值给返过去，那这样的话呢，咱们经过处理之后，我们点VR一下，点VR一下咱们拿到一个stream对吧，那这个呢，应该是一个Jason stream。好了，那么这里面放的什么，咱们建对象啊，对吧，然后您去调用咱们Jason它里面的一个方法对吧？来把咱们这个数据呢，给打印输入出来啊给打印输入出来好了，那这块呢，是我们现在啊从卡卡里读数据，然后读完数据之后呢，那我现在呢，对咱们数据做了一个简单的处理，怎么处理的呢？对吧，把这日期拿到，拿到日期之后呢，我现在求出了咱们天以及咱们的小时，那就是天以及咱们的小时，好，那这样的话，咱们现在可以运行一下，呃，我如果要连这个网的话呢，比较慢。

把这个呢，关掉了。多少？锻炼一下。我这个断开对吧，然后呢，我现在呢，去运行咱们这程序来看效果啊来看效果。好了，那么咱们现在呢，是不是可以生成数据了呀？来，在这里我们的模拟生成数据。Success对吧，那么这边呢，这个数据大家看这边这个数据呢，那应该呢，是不是也过来了呀，对吧，好了，那咱现在来把这个停一下，对吧，把它停一下之后呢，咱们得看一看啊，我们这个数据生成的到有没有问题，到有没有问题，首先呢，咱看D题有没有。

然后进下来，除了DT之外呢，应该还有什么，是不是还有HR呀，对，在咱们这里边是不是HR也有了呀，对吧？哎，所以说呢，你看在咱们现在经过我这个处理之后，我有什么DT，对吧，今天HR呢也有了，那到时候咱们有了这些数据之后，我是不是就可以知道咱们每一天每一个小时啊，它的一个访问量呀，对吧，这里到时候还需要什么呢？什么mid啊这些东西。对吧，那这个呢，就是什么呢？这个这就是关于我们现在对吧，这个第一个我日活的一个第一个功能对吧，但第一功能呢，目前咱现在其实就是做做在这儿了，对吧，干什么呢？通过我们Spark催敏对吧来去消费卡法数据对吧，那么像绿色标记的这些咱们前面都已经做过了，咱们主要是什么编写一个SPA推对吧这样的一个我们这个类对它的个数据呢，来做一个消费对吧，来做一个消费对吧，那么在咱们这个消费的时候，大家注意，首先呢，调用我们的工具类来拿到我们的stream对吧，然后接下来对吧，那么咱们拿到之后呢，对吧，对它做了一个结构的转换对吧，原来拿到的一个record，对，这个record是不是就咱们当前从卡卡里面读到的数据，它封装的是咱们consumer record对象，但是我们现在如果真获取我们这个值的话，是不是只需要拿到value就可以了，所以说呢，我们现在通过调度value方法把这个东西拿到，然后接下来咱们现在呢，把这个接。

字符串转换成层对象，然后呢，去获取我们接对象它的一个短时间戳，然后通过时间戳呢，创建咱们的日期对象，调用我们这个c data for made，将我们这日期来转换成我们指定格式的这个字串，然后接下来这字串做一个我们这个啊spli的分割，那么分割之后呢，得到两个内容，一个呢是我们这个小时对，一个呢是我们的小时，我们给DT重新复过取，那么另外一个呢，是啊，一个是天对吧？那么另外一个呢，是我们先小时对吧？另是小时对，然后最终把这个东西给它打过去，然后咱们做一个打印输出对吧？这是这个过程对吧？那么来到我们现在这里啊同学们，咱们现在呢，这个过程应该这样的，对吧？现在这些东西都有了，然后呢，我现在SPA是不是从这里开始读数据了，对吧，那么读数据之后，咱们现在对数据呢进行处理，我目前处理方式只是打印输出了，对吧，我对大家明显的要求呢，也是这个。

对吧，就不用做太多啊，不用做太多，太多我也查不过来是不是，然后你们就做到这就可以，对吧，就做这也可以，那么今天呢，咱们主要给大家介绍内容，大家需要下去做的东西，首先呢，第一个把我们这个search里边它的一个咱们的索引模板，对还有别名这块呢，也稍微练一下，走一遍，然后接下来咱们是不是主要做了一些我们这个准备工作呀，其实我们写了几个工具类，一个呢是我们以联search的工具类，但现在其实以联search呢，这个东西算不上工具类，虽然说叫工具类啊，其实它里边的方法只是让你对吧去熟练我们的S设置，通过咱们这个客户端去咱们的操作我们ES，对吧，通过咱们这个SKY代码去操作它，对吧，然后呢，我们自己写了一个读取我们配置文件的工具类，写了一个我们从卡卡里面获取数据的工具类，写了一个获取我们的ready客户端的工具类，把这些工具类都做好之后呢，咱们做了其中第一个功能。这些工。

这是干什么的，这个功能就是从卡卡读数据对吧，那么这里它会用到我们这个工具类对吧？其实我们这个从卡卡读数据的工具类，咱们到这儿是不是就用上了呀，对吧？哎，所以说你现在通过这个程序呢，也能测试出来你的工具类对吧，写的是否正确对吧？如果做的快的话，那么你可以考虑再往下看一看，怎么样去帮咱们驱虫啊，怎么样帮他驱虫啊，然后等咱们这个后天来的时候，那大家呢，也是提前准备好，我到时候看的是你们这块的东西。

对，就你们运行你们在这里对吧，能把这个数出来这就OK了，好那么咱们今。