025_尚硅谷大数据技术_用户行为数据分析Flink项目_广告点击量统计（一）_基本代码实现

来我们讲另外一个市场营销相关的指标，那就是页面广告分析，大家知道在这个电商网站，或者说其他的一些网站也一样啊，啊大家知道在这个市场营销的指标里边，除了这个不同的推广渠道统计之外，那另外还有一个就是自己页面上的一个广告投放，对吧？哎，那么这个广告投放其实分成两类，就是一种是我当前自己的页面上给自己的一些产品做推广，做一些呃做做做一些，相当于这个直接打广告去去做这个营销，对吧？哎，那么这是考虑到就是对自己，呃推广内容的一个哪些地方啊，推广的效果更好，可能要去做一个统计总结，那另外呢，而是有可能我们一些主要的一些核心的页面，是别人在我们的页面上要去推，就是要去做一个推广产品，要去打广告，对吧，那大家会想到对于我们当前网站上的一些页面，是不是应该涉及到一个广告收费的问题啊，计价的问题，对不对？

那你按照这个这个广告按照什么去计价呢。直接拍脑袋响吗？啊，因为大家想到不同的页面，它的热门度如果不一样的话，是不是相当于，哎，我们这里边的这个广告被点击的可能性也就不一样啊，这就像我们这个传统行业里边，呃，电视上打广告，大家知道都会划分这个，呃，就是当前我们这个时间段对吧，黄金时间段，它的那个收费就就比较高，那这个黄金时间是按照什么来定义的呢？传统行业的话，那是看那个是不是收视率啊，对吧，如果要是人看的多，那是不是相当于我这个时间段比较受欢迎，那你要打广告费用就高对吧？啊，所以在有一些我们这个网站当中，对于这个呃，其他一些客户去投放的广告去做计价，计价定位的时候，定价的时候其实也是按照当前页面这个广告的点击量来考虑的，对吧，就是当前如果这个点击量越高，那是不是相当于我这个页面就非常的火爆，对吧？啊，就是对应来讲的话，相对应就相当于我那个收视率高嘛，那当前应该就是计价要高一点啊，所以。

页面广告的这个统计分析，在有一些场景下，对于这个市场营销策略的调整还是比较重要的，总要统计出一个数来，对吧？啊，所以接下来我们来做一个这个需求的实现啊，啊，那对于这个广告统计来讲，最简单的也是最重要的，那就是直接统计当前页面广告的点击量，大家想这个就很简单了啊，就是来一个点击数据，我直接就做一个统计，Count加一不就完了吗？啊这里边我们接下来要提出的一个需求呢，呃，是要做一个精细化的划分，就是按照什么来调整，按照什么来分组统计呢？按照用户不同的地理位置进行一个划分。因为大家想到就是我当前不同的广告，不同的商品啊，它被点击的这个程度啊，它的这个热门程度，是不是有时候确实是跟这个不同的地段会有关系呢。

比方说大家想到这个，呃，不同的季节啊，呃，然后我们整个这个中国幅员辽阔，南北差异其实很大的，你像到冬天的时候，如果大家是在这个东北或者说在北京啊，北方的这些地区的话，其实冬天很冷，你如果一个一个商品这个广告啊，是这个羽绒服对吧？呃，这个大大棉袄对吧，这这样的一个推荐的话，这样的一个推广的话，那可能会比较受欢迎，但是你想想，如果说要是在南方，在海南对吧，在三亚啊，那大家想这个即使是冬天，它温度可能还十几度呢，根本用不到羽绒服什么的东西，对吧？啊，所以说这个这个地域的差别其实还是比较明显的。所以我们当前这个需求就是开一个窗口进行一小时数据量的统计，然后我们统计的是当前啊，就是每一个省份啊，按照省份划分，统计当前每一个广告的点击量到底是多少啊，这是这样的一个基本需求，所以接下来我们就来实现一下这个需求了。

还是在代码里边，接下来在当前模块下啊，直接新建一个class。当前的这个模块我们就叫做，呃，这这个其实是应该应该叫做一个广告的分析，对吧，所以我叫做ad，大家知道ad就是广告的那个缩写啊ad，呃，统计statistics。呃，然后这个我我是基于这个省份去做划分的啊by province做一个这样的一个定义。哦，这个province好像写错了是吧，我们把这个还是修改一下啊。当前这个province啊，加一个N。这是我们当前想要的这个数据，呃，当前想要的这个类啊，那里边自然还是PSVM，整个这个流程是一样的，Throws exception。

呃，Throw一个exception，然后下边首先还是stream execution environment，我们先把这个环境先创建出来，叫做env里边这个env，先把这个时间语义啊，哎，那大家想一下，当前这个时间语义正常来讲，我肯定应该还是按照这个事件时间来处理，对吧，应该是我的数据日志里边应该是有这个时间字段的啊，那另外把这个全局的并行度直接先设成设成一。那另外我们就想到了当前这个广告的点击数据，我们那个user behavior里边也没有啊，啊，那没有这个数据怎么办呢？那就只能是自己去创建这个测试数据了，对吧？啊，那当然现在我们不需要再去做这样的一个自定义测试数据源的创建了，我们直接拿这个测试文件作为这个数据源就可以了，我们在data里边。

找到这里边有一个文件直接就叫做ad click log.csv啊，所以大家看到这就是我们自己生成的一些测试数据啊，我直接把这个放放在这个resources下边哦，那我们先看一下吧，这个CSV文件又是逗号分割的啊，这是明显是做过ETL之后的那个形形式对吧？那这里面大家看一下里边的字段有哪些呢？哎，我们直观来看的话，前面好像是两个ID对吧？那所以这样的ID应该是什么ID啊，首先大家想，既然是用户对广告的一个一个行为嘛，那我们这里统计的其实就是对哪个用户点击了哪个广告，所以是不是应该是user ID和广告ID啊呃，A didd对吧？呃，那这里边就没有用户具体的行为，我直接把它滤掉了，ETL掉了，因为大家想这里边是不是统计的都是点击行为对吧？对于广告你没别的操作了嘛，只有一个点击，然后另外还有就是哎，大家看接下来这个北京，北京为为什么两个北京呢？啊对，就是省和市对不对，对应的这个具体的细节的这个地理位置都有啊呃呃，北京北京对吧，广东广州啊，大家想这个我们现在都基于那个手机APP嘛，所以说统计这个地理位置其实很简单对不对，对吧，只要我们那边开启了这个啊，对应的这个定位服务啊，很容易都可以收集到对应的这个定位信息，所以这些信息都非常重要。

呃，那后面最后还有一个字段，就是时间戳啊，这个大家一看就知道是怎么回事啊，所以接下来我们就来定义一下了。当前是不是首先还是得在并下边。定义对应的一个port类啊，哎，大家看这个这个数据包装起来应该长什么样呢？呃，我把它叫做呃ad，就是当前的这个广告对吧，广告点击事件吧，Ad click event。

然后接下来我想要的这个字段，其实首先就是一个长整型的user ID，然后是一个还是长整形的a did对吧，接下来接下来是那个province和呃，这个城市对吧，省份和城市，那我就定义成string类型的province，以及string类型的city。然后最后还有一个长整型的时间戳time，呃，这就是我们一开始做的一个定义对吧？还是空餐的构造器先定义出来，带参数的也定义出来，呃，那后边我们还有这个get set，对吧，全部定义出来，最后。把这个to string也写出来，哎，非常快速的，我们把这个po类啊先生成好，然后接下来大家想到，那我们最后想要得到的那个统计结果，是不是也应该相对来讲有一个包装啊，按之前我们的做法对吧？我们最后想要知道的是不是也是类似，就是某一个省份，然后当前可能我开了窗口嘛，所以我要输出一下当前的window and对吧，当前的窗口又是哪个，然后最后输出一个当前省份里边，哎，所有这个用户点击的，点击广告的那个数量有多少对吧？来做一个count，求求出的那个count值输出，所以当前我在定义一个port类。

这里面就定义成比方说就叫ad count，对吧，呃，View。By province。Province，就像我们一开始那个item will count一样啊，只不过我们这里面是一个a count啊。呃，这里边接下来我们想要的字段，其实就是首先应该是province对吧，这是我们的那个相当于是key了啊，关键关键字，然后接下来是window and对吧？哎，我还是把它包成一个string吧，看的清楚一点啊，Window and，最后是一个哎，一个count数量private长整型的count，所以还是把当前的这个constructor空餐的和带参数的先创建出来。

然后接下来还有这个诶。在这对吧。Get也都创建出来，最后还有一个to string，好，这就是我们想要的基本的数据，数据类型啊，数据结构先定义好，呃，然后接下来我们想做的事情当然就是。是不是先读取数据了对吧？首先啊，从文件中读取数据，呃，那这个过程大家想到一开始我们都是直接从那个文件里边，就是全量数据直接就拿出来了是吧？那我们现在一般都是要呃不不是全量数据，就是说呃直接写死那个全全路径对吧，直接把这个文件读取出来，那我们现在的做法一般都是是不是用那个resource啊对吧，用反射获取到当前的这个类的。

呃，这个class，然后我们get resource，给一个相对路径就搞定了，对吧，Ad，那个叫click.c这个叫ad click log，对吧，点CSV。得到一个resource，然后接下来是要得到这个。呃，或者我们这个叫一个其他的名字啊，我们叫叫ad click event stream对吧，这个大家看的更加的明确一点啊，就是点击事件广告，点击事件的那个流，那是event前面我们直接定义出来啊。Data stream。里边的数据类型读进来是不是应该想转换成一个ad click event呀，接下来就是env，首先read text file，这里边我们是resource.get pass先从路径里边读出来，然后接下来是不是直接做一个mapb转换每一行啊，又来了对吧？得到的这个首先是不是做切分得到一个字符串类型的数组啊，啊，我们把这个还是叫做fields，等于每一行做一个split，现在是CSV文件逗号分割对吧？然后后边是不是直接return一个new ad click event里边的字段首先是长整形的。

呃，User ID对吧，New一个长整形，这里边要的是FIELDS0，然后后边你一个长整形要的是FIELD1，因为大家知道后一个是aid嘛，都一样对吧，然后接下来是省份和城市啊，那这两个是不是都是字符串啊，呃，所以FIELDS2和FIELDS3，最后还有一个长整形的时间戳，长整形fields。

四诶，这样就搞定了对吧？这就是我们想要的这个数据类型啊哎，那当然了，尽量的靠近数据源，我们是不是要分配时间戳和watermark呀，对吧，已经定义出来了，那接下来又一个，哎，问题又来了，当前的这个数据到底是升序还是乱序呢？那还是得看一眼了是吧，我们看一眼这里的数据啊，5658000，然后6580601分钟之后对吧？啊120，因为测试数据嘛，大家看都是一分钟一分钟的是吧？180240，然后后面啊，这里面1234都是升序对吧？啊这个就看的很明显，所以这里边直接又一个啊sending time呃，Extract啊，这个提取器里边我们提练的是element.get time sample，注意这里边的时间戳，这是一个秒对吧？哎，所以那我们现在应该还有乘以1000，把它提炼出来。

这就是前面做的这个基本操作啊。然后后续当然就是要基于是不是基于省份分组啊，然后是不是开窗聚合，开窗聚合对吧？啊，那接下来我们同样还是基于这个ad click stream，首先啊，我直接K对吧，当前好像也没有什么需要filter的，对吧，然后那我就直接K了K。是不是按照那个provi做一个P啊，诶，那这里边我可以直接写方法引用对吧，大家知道如果方法引用的话就可以，呃，就是后边的话就不是他类型的啊，所以这里边我用这个ad click event，然后get。Province是不是就可以了，对吧？直接调这个方法引用，那么K外之后开窗，我们现在要的是一小时的时间窗口，当然了，因为数据比较少，所以说我们还是定一个滑动窗口给大家看的，呃，输出的东西多一点，对吧？因为你如果直接统计一小时滚动的话，数据就很少啊，呃，窗口就很少，所以接下来我们每隔五分钟滑动一次吧。

我们这里边是，呃，这个啊，直接写在下边，接下来继续做这个开窗，后边这里边写一个注释定义划窗。五分钟输出一次，然后接下来就是聚合了，那家知道这个聚合既然又是来一个，就应该count count一次对不对，然后后边是不是又要包装，连同这个window信息包装在一起啊，那又是aggregate，对吧？啊，又是之前的流程又复习一遍，你有一个比方说我这个叫a count a j，这是不是应该是一个增量聚合函数啊，然后后边再去用一个全窗口函数，对吧？Ad count result。

把它定义出来，呃，这里我们可以给一个当前的这一个结，聚合的结果，我直接就叫做ad count result stream。最后我可以把它做一个打印输出，不要忘记还有这个XQ的执行起来，对吧？啊，当前的这个job name叫做ad countt job by province。Job，这就是整体的处理流程啊，当然这这两个自定义的函数类也非常简单，是不是啊？那这里边我们就还是再来实现一遍吧，Static class a count a。他要去实现的接口aggregate function对吧？Aggregate function里边的类型in ACC out，这个大家其实都已经都已经太熟悉了是吧？当前输入的port类型是不是就是当前的input类型啊？呃，A click event对吧？然后接下来是不是都是长整形啊，因为它不需要包装嘛，只要拿到那个count数量就完事了，然后接下来实现这四个方法，Create的时候0L at的时候accumulator，加一，后边get result的时候accumulator对吧？最后墨的时候A加B。

哎，这这个过程大家都都觉得非常熟练了是吧？Public static，后面我们再来实现一个自定义的window function a count result对吧？它需要去实现的是一个window function，当然大家想写那个process window function也行，是吧，前面我都已经做过了啊，呃，然后它的类型是in out key和。

最后的那个w window，所以当前它的输入是前面增量聚合的输出，长整型，它的输出是不是我们最后要的那个定义好的那个po类啊，Ad count view by province对吧？呃，然后最后当前的这个K类型是不是string啊，因为我们前面用的是。对，前面用的是方法引用做的这个KBY，我把这个换行换过来啊，方法用，那么这里面相当于是传的是一个k left得到的，本来这个key是什么类型就是什么类型，对吧？哎，所以这里边非常简单，Province本来是string类型嘛，这里就是pro province就是这里的key就是string类型，最后还有window类型time window，好，那最后我们要实现的是一个apply方法，那这里边的这个apply方法的话，当然就是说还是定义一个window and，对吧。

呃，就是大家知道这里边的这个string，这是不是就是我们当前的那个province啊，对吧，我直接把它这个重重新写一下啊，这个就叫province，然后接下来要的这个window and呢，那还是new一个time time。把这个引入对吧？然后里边直接用window.get and，然后后面我们再to string就完事了，对吧？这是window and，后面我们还有一个count数量，那是不是又是input呀，Input点点next直接取出来就一个数嘛，最后要输出的时候是不是out.connect你有一个包装成想要的这个输出的步骤类型ad countt view byvince，里边就是当前的province，然后window and还有一个count，这是不是就是我们最终输出的结果啊？啊，所以整体流程还是一样是吧，就是你定义好只呃，这里面还有一个问题，就是object是不是要改一下呀，当前输出的那个叫ad count。

这个叫ad cant view by province对吧？哎，我们要做这样的一个处理，所以。主体整体来讲就是首先把这个数据读取出来啊，转换成自己想要的这个po类型，然后接下来我们就是基于某个字段定义好的字段去做分组开窗，然后做一个聚合统计，我们这里边的聚合还是增量聚合，来一个就加一，另外得到的结果呢，结合窗口信息，再包装成一个想要的步骤类型就可以了。大家看一下这个输出结果怎么样啊？运行一下，看一下效果。大家看一下哦，当前已经输出了，大家看我们当前的这个输出就应该是五分钟输出一个结果对吧？呃，09:05的时候，当然这个因为是有不同的省份嘛，所以这里边你看到这个是有多条输出的啊，比方说这个北京只有两个点击啊，上海一个对吧？呃，这个广东啊，呃，这里边你看零五分一个一个窗口，十分又有一个窗口对吧，15分又有一个窗口，五分钟滑动一次，然后得到这个统计量啊。

各自都都不一样对吧，大家看中间有一段时间，这个北京的这个数量非常大的啊，120多次啊，这就是我们按照省份统计广告点击量得到的一个结果。