II_项目_电商用户行为分析/064_尚硅谷_电商用户行为分析_带黑名单的广告点击统计（上）

接下来我们分析在这个市场营销商业指标里面的另外一个比较常见的指标，那就是跟广告相关的这个指标，大家会想到在这个电商网站里边，几乎所有的就是所所有的这个网页啊，上面其实很多角落，很多边边角角都会充分利用起来，那最常见的这种利用页面的方式就是给用户推送一些广告，那这些广告大家会发现现在其实很多广告都是要讲究。精准推推送精准投放的对吧？啊，因为这个你如果要是一个广告是真的是它的那个受众人群，你给他推送出来的话，这个转化率是非常非常高的，但是你如果要是不知道该给什么样的人推的话啊，那你说你在这个不管是什么时间吧，你给一个住在海南岛的人去使劲推这个羽绒服，推这个棉大衣，呃，显然这个销量不会很好，对吧，没有人对这个东西感兴趣啊，或者说呃，你你不停的对这个小孩子或者说年轻人去推这个什么养生保健对吧，什么枸杞茶之类的东西，那肯定没有人去点嘛，所以啊，对于我们电商而言啊，这个一方面也是跟其实跟这个用户的就是画像对吧，这种分析相关，跟推荐相关，另外一方面呢，它也涉及到就是用户有多少的点击量，我们到时候给他做精准投放之后，怎么样去按这个页面去算钱，对吧，广告是有这样的一个营销方面的考虑的。

那从这个网站平台的角度讲呢，我们需要知道这个页面上到底这个广告它有多受欢迎，就是这个页面到底有多受欢迎，这个点击量到底有多少，对吧？然后我根据这个去做一个定价，做一个推广和营销策略的调整，所以在这个过程当中，其实我们要做的一个统计就是把用户点击广告的行为筛选出来，然后啊，做一个聚合计算，对不对啊，在这个过程当然可能也会有一些时间窗口的要求，比方说我们统计一个小时之内的广告点击量，然后实时更新，比方说五秒更新一次，几秒更新一次。另外还有一个就是常见的需求，就是可以根据不同的类型对用户的点击做划分，比方说呃，我们说不同的地域是不是相当于这个点击量，应该是会有不同的考虑啊，那你比方说这个，呃，在广东在海南。

那有一些页面的点击可能就会比较频繁，对吧，有一些页面点击，那相对来讲就不那么受欢迎，所以在不同地方广告的投放可能会有不同的定价策略，那这里边我们就实现这样一个，首先先实现一个简单的页面广告点击量的统计，这里边我们要按照不同省份去做一个划分。好，那所以接下来我们就直接在代码里边去做实现吧，呃，我们在这个market analysis，直接在这个包下边新建一个object，这个就叫做呃，统计信息啊，Stat。

Statistics by jail，呃，大家可能知道这个geography就是地理位置对吧？呃，跟跟这个地理位置相关的一个广告点击量的一个统计，我们先把这个创建出来main函数。那么在这里边我们的数据用什么数据呢？同样正常来讲，对广告的点击也应该放到买点日志里边去，对不对啊，这里边user behavior同样没有广告相关的点击数据，我们这里边有一个文件，呃，对，ADD click log这个文件就表示了我们能用的这个数据啊，Resource被删了是吧？

好，我们还是把它扭出来吧。Resources。哦，这个大家看应该当前的状态就是已经被mark成那个resource文件了，对吧？好，呃，然后我们把这个对应的文件copy到这个里边来，大家先看一下这个文件里边我们存储的这个数据是什么样子的呢？其实这个比较简单，这已经是做过这个ETL提取过的，对吧？所以它其实就记录了每一个用户对每一个广告的点击行为，这里边我们收集到的数据是，那就应该是一个用户是不是点了一个广告啊，所以是有一个user ID，然后还有一个对一个广告的ID，然后接下来是不是就是，哎，这里边当然有地理位置，那就应该是省份和城市，对，最后还有一个时间戳啊，这就是我们要处理的这个数据类型，那这里边还是先定义出来样例类吧，对吧，输入的。

嗯。广告点击事件样例类。His class。Click event。呃，这里边user ID这个是一个浪类型，然后广告ID对吧，ADDID，另外还有province city street，还有一个time STEM浪了一些，对，这样就把它已经。定义好，另外有输入就有输出，我们最后要输出的一个结果就是根据省份划分出来的，最后广告点击量的一个一个聚合统计，对不对，一个count值，所以这个也包装成一个样例类吧，按照省份。

统计的输出结果样例类。呃，这就叫count by province，呃，那这个我们可能关心的就是说到时候还要开时间窗口对不对？呃，我们可能关心的就是哪一个省的，然后在哪个时间窗口里边的，到底统计出来多少多少次点击对不对？所以这里边我们可能需要有一个window and long类型，然后province string类型，最后还有一个count long类型，对吧，我们先把它定义好。那接下来这个程序的主体流程，这个就这个就非常简单了。

大家会想到是不是跟我们之前实现的那些都差不多啊，所以这里边先是环境先拿到对吧，Get environment，然后接下来还是啊，我们先设置这个时间语义，我们是不是要time啊，对吧？因为当前我们还是有一个那个时间戳进来的啊，然后另外我们方便结果输出的话，可以把这个并行度先全局设成一。然后接下来就可以读数据了啊，我把这个定义一个resource。Get class大家还记得这种方式对吧？相对路径的这种resource，我们把它当前的目录copy下来，写到写在这里。然后接下来我定义一个ADD event stream。

呃，当然这里边啊，我们先定义出来，后面再做单独处理，对吧？先把它这个单独的事件提取出来，后边基于它再做进一步的处理，所以这里边就是env，对，Read text file，这里边我们要传的是resource.get pass。读进来之后是不是要做一个map呀？呃，对每一个data都是不是要做切分啊，这个大家都熟啊。它的是按什么隔开的啊，逗号隔开CSV文件对吧？然后最后要把它包装成一个ad click event，呃，那么这里边user ID就是datara 0tri之后出了对吧？啊，当然下面的那个I就是a dad也是一样图壤对吧？呃，然后接下来是provin和city，那那两个就都是都是string，所以直接tri一下完事。

Date。三对吧。最后还有一个time stamp，它是四拿出来之后要。要把它转换成长整型图了，对吧？哎，这就是我们一开始的一个转换的过程，所以这里边是读取数据并转换成。呃，Ad click压梨类的这个类型对吧？Event类型好已经读进来之后，接下来呃，我们要做的，大家想想接下来要做什么呢？诶注意啊，我们这里边设置的是事件时间，我们只是把它map成了对应的这个样一类的类型，还没有指定时间戳和watermark对不对？诶这里面那指定这个时间戳watermark的时候，可能就得去具体问题具体分析了啊来我们来看一看这个时间。

肯定是用这个字段做时间戳对不对啊，然后我们看一下80060120啊，这个看起来这个比较好，对不对啊，010203啊，这个是升序的数据啊，所以这个数据的话，我们就直接啊，Sign啊，Sending time steps对吧，直接把这个定义好就好，那这里边提取的是time stamp去乘以1000L，因为其实我们已经看到了这个位数比较少对吧？呃，这个十位显然表示的这是一个秒，以秒做单位的一个时间戳，先把它定义好，然后接下来就可以是不是根据省份做分组，然后开窗聚合了，哎，所以这里边我们定义一个ad count stream。

它就是前面的event stream基础上要去做，对，首先是不是根据province做分组啊，K by province啊，当然如果我们对这个地理位置的需求是你要把CT也算上的话，那就把CT也包括进来，对不对？呃，这里边大家注意K的过程当中。有几种不同的替代方法，这个大家知道对吧。在这里边，如果我们要有多个K的时候，怎么做呢？首先大家看如果是给一个int，或者给一个这个string字段的话，是不是可以给多个值啊，但是注意这种传入之后，后边K的类型就是一个Java，他对那那如果说我们就用这种方式，这个是不是得到K的类型就是当天province的类型啊，那如果我们想要用多个K怎么办呢？

哦，那是不是可以包成元组啊，对吧，就不可以把它包成一个元组作为K就可以了，大家把这些都可以做一些具体的实现啊，然后接下来我们就开窗开window。一个小时统计，那这个就还是探点二一，那么可能是滑动窗口的话，还是五秒钟滑动一次SECONDS5对吧，这都是常规操作，然后接下来是不是要做聚合啊啊其实这个非常简单，就是KBY这个province之后，接下来是不是也是来一个数据就聚合一次来来一个数据加一，来一个数据就加一，那大家联想起来是什么呀。是不是就是我们之前那个就count a j是不是就干了这个事情啊，做一个预聚盒对不对，最后再包装成一个count by province这样的一个输出，这是不是就是我们之前做的那种事情啊，所以这个就完全是一个复习啊，Aggregate还是啊，你有一个这里叫ad count agg。

然后后面我们用一个ad count result在输出啊，这样就可以得到它的结果了，对吧？当然我们最后把这个count stream打印输出，看一看最终的结果是什么样子的。最后加上执行对吧，给一个drop的名字叫ad。Statistics job啊啊，这是我们的主体流程，然后下边具体的实现也是非常简单，Class a count a j，它要去实现一个aggregate function的一个接口，那么这里边的类型是input，然后累加器，对，还有一个output，那么这里的input是。

是不是就是我们定义好的那个样例类类型啊，Click even的对吧，然后累加器类型就是对，就是浪，它就是一个技术嘛，那么这里的输出是不是也是浪啊，因为我们说前面语句和函数的输出就是最后的输入对吧？啊，所以这里面还不用包装成那个样力类呢，好，然后这里边的实现。哎呀，这个太熟悉了，对吧，现在大家看这个都都不用花时间直接加一来一个就加一个对吧，这个是我们的自定义语句和函数。这里边初始是不是零啊对吧，然后get result的时候直接返回累加器，这是我们的这个聚合的状态嘛，然后这里边如果要是叠加的话，A加B对吧，然后接下来自定义窗口处理函数。

Glass，呃，这里边我们那个名字叫做ad count result，他要去继承一个，诶是是kid吗？大家还记得吗？对，这里边我们是不是直接去直接可以去实现一个window function就可以了呀，对吧，当然我也可以去实现，呃，就是对应的这个process window function啊，这里边其实没必要啊，我们直接就用一个window function包装成一个样例类是不是就可以了，直接输出就可以了，Window方式。而且大家还记得就是我们当时。

就是给大家看过，它是一个skyla的tra，一个特征对吧？而上面的这个是一个Java的interface，这个大家要搞清楚啊，这里边要传的数据类型是四个，又是这个来了input output key和window，所以它的input就是上边的输出就是law，然后output就是对样例类count by province。另外K是什么？大家想一下K是不是按照province分的啊，所以这里边直接就是string，对不对？然后最后还有一个window是time window。对。然后把这些实现之后，上面都不报错了，接下来我们就要实现它里边的一个apply方法，这个apply方法也非常的简单，是不是直接out.collect我们把它包成一个count by province这个样例类啊，里边要的这个数据是一个应该是一个window and，对吧？然后是一个呃，还有什么来着，还有一个province，还有一个这个po对吧？啊，这里如果大家觉得这个window and，这个这个long时间戳不太好看的话，我是不是也可以把它换成一个string，就可读性更强的，呃，年月日那样的形式啊，对吧？这个大家都已经熟悉怎么去做了啊好，这里边我们就把它包装一下，你有一个time STEM。

这里面要传进来的是。传进来的是什么呢？是不是就是window点，因为window and嘛，是不是window.get呀，呃，当然这里边我们在对这个在涂string对吧，把它转换出来，这里可能需要把这个。

好。然后接下来后面一个是后面一个是当前的province对吧，那province是不是就是就是K对，然后还有一个还有一个count值，Count值在哪里，是不是就在input里啊input。点这是个迭代器对吧，然后next直接把它取出来就可以了啊，当然这里边其实没有没有必要这么去去写对吧，你可以把这个删掉，直接to string也是一样的，所以这里边我们就把它已经包装好，输出最后的结果了啊，我们可以看一看这个现在的这个状态是什么样吧。看一下执行结果啊，这里边已经正常输出，我们这里边统计结果了，对不对啊，因为是五分钟一个窗口，所以这个输出的数据非常多啊，就每一个窗口里面统计多少个都有，呃，诶，然后大家在看的时候其实发现。

北京的这个这个统计量好像挺大的，对不对，好像这个数挺大的啊，如果我们要是仔细分析这里边的数据的话，其实会发现。大家看这个这个北京的这个数据为什么很大呢，大家发现是。同一个人点同一个。同一个广告对不对，然后是不是在短时间内刷了很多次啊，大家看是不是这样的一个行为啊，所以在这种状况状况下，这有点就像我们的那种刷单行为了，我们是不是其实想把这样的人要过滤出去啊，你不要因为我们要统计这个信息，结果有些人他就他就在那里狂点，呃，或者说甚至是写了程序之后，哎，直接在那里刷这个点击量，如果我们把这个都统计进去的话，那其实这个大量的是一些无效数据，而且我们会认为这样的用户是有风险的，我们其实应该把它加入到黑名单，对不对啊，今天。

可能就是今天之内，接下来他的所有的点击操作，我们可能就呃都无效了，都都认为他这个点击不再去作为我们的统计了，那或者说我们把他加到黑名单之后，还有一些后续的操作，比方说去发短信通知啊，或者说呃，做一些账号的管理，这个权限限制啊，这这些就是我们后续业务逻辑可以定义的一些事情，所以我们这里边可能有一个比较直观的需求，就是说对于这种大量刷单的行为，大量刷点击量的行为，可能需要在这个过程当中，是不是要把它过滤出来啊，哎，就是不要把这种刷单的数据统计在我们最后的结果里边，而要在前面把它过滤掉，而且呢，在过滤过程当中，一旦发现有人今天比方说已经连续刷了100次了，对吧，对同一个广告刷了100次了，因为页面上可以有不同的广告嘛，他如果这个人就喜欢就喜欢乱点点了一。

S，我们觉得也正常对吧，你不排除有些人他就这个这就好这口，呃，所以我们的我们的观点是他如果就点一个广告连续的点对吧，然后点了100次以上，那我们可能就他接下来的这个行为就都无效，都不统计他的行为了，而且会把这个用户加入到一个黑名单里面去，哎，这是我们的一个需求，那当然了，这个你不应该把人，人家之前如果某一天有过这个这种对某一个广告点了100次的行为，你不能说以后一直把人放在黑名单里面，对不对啊，所以我们可能做一个这个时间限制，就是比方说以自然日作为一个标准。今天之内你你连续点了100次之后加入黑名单，然后你今天的这个点击行为就无效了，而从明天开始呢，那这个你的点击就又有效了，对吧？就相当于是每天的零点，晚上12点的时候，相当于把这个状态清空，第二天统计的过程当中，就相当于又可以重新统计了，但是黑名单的那个操作的话，我们不删，就是相当于我们可以有自定义的后续操作，呃，你是对他权限做限制，还是说给他一个发送一个通知，这是我们可以去定义出来的行为，好，那大家想一想，接下来我们怎么样去做这件事情呢？

这个过程其实呃，既然是要做这个黑名单，然后做相当于是要做一个过滤了，对这个输入的这个数据是不是要做做一个过滤啊，呃，对，所以现在我们接下来是要干什么呢？是不是得记录一个用户对同一个广告的点击量啊啊所以在前面其实我们比较感兴趣的一个一个东西是什么呢？其实是要对拿到当前用户对同一个广告点击，到底到底已经点击了多少次。

那我们是不是可以把这个东西当成状态对保存起来，所以我们可以用状态编程的思路把它解决掉啊，那另外就是说像我们所所谓的这个每天到12点整点的时候要清空，要重新开始记，那这个这个任务怎么样去实现呢？对，那肯定就是定时器了，对不对，而且大家会想到我要设定一个相当于是处理时间的定时器，对吧？啊，我是按照这个系统时间来算的，然后设置每天的12点钟把我们的状态清空就可以了，所以整个来讲又要用到状态编程，又要用到这个定时器时间相关的操作啊，那我们怎么去做呢？是不是直接应该上大招了啊，直接到时候process方式，直接一个点process，自定义一个这个process方式，把它所有的任务就都可以搞定了，接下来我们就来实现啊。

呃，首先这个既然是黑名单嘛，那肯定是应该有一个对应的，呃，输出这个报警信息的对不对，黑名单的一个报警信息，那这里边我们还是定义出来啊，呃，输出的黑名单报警信息case class。这个我们叫做呃，Black list warning吧。那这里面我们关心的信息可能就是user ID对吧，到底是哪个用户进了黑名单了，呃，然后另外还得知道他到底点了哪个广告点了，呃，比方说我们定义超过一百四以上，所以这里边aid。

诶，刚才写错了啊UID是一个long类型对不对，然后另外还应该有一个比方说报警的信息message，这是一个string。先定义好这个，然后接下来大家说我们做这个过滤是在哪一步去做过滤呢。是不是应该在已经得到这个even string之后，然后在下边是不是就应该。对，提前先处理一下，把已经进入黑名单的那些用呃，那些用户接下来的点击行为，是不是就应该把它都剔除出去了呀？啊，那如果我们保存了那个状态的话，其实是可以就判断那个状态如果大于我们设定好的那个上限值，那是不是就应该把接下来的数据都过滤掉，你不要再再继续往下传了，对吧？所以其实是在前边去处理的。这里边我们自定义process function。

过滤，呃，大量刷点击的行为，所以这里面我们可以定义一个future black list STEM，呃，它当然是会基于ad event stream做什么操作呢？首先是不是应该先做一个，那大家想到这里边我可以直接那直接process的话，那实现的就是一个基于data stream的process function，对吧？但大家会想到这里面我们是基于什么去考虑的呢？是不是基于同一个user点了同一个广告，我们要保存的是这个状态，对吧？所以不同的用户是不是应该保存不同的状态，不同的用户对不同广告的点击量是不是也应该保存不同的状态的呀？怎么样可以实现这样不同状态的划分？

那是不是我们可以用一个key的process function，就一开始先应该做key by，根据什么字段做key呢？ID，对，是不是就是用户ID和广告ID啊，哎，这里面就涉及到两个筛选条件，呃，这个两个K了，两个K怎么实现？对，如果我们想用一个元组去实现，当然我们可以直接这里边传字段对吧？大家看我是不是可以直接user ID dad这样传是不是就可可以啊，只不过这时拿到的那个K的类型就变成了一个Java temple类型，我们如果不想要这样的话，那可以稍微麻烦一点，对，把它转换成一个元组对吧？那这里边就是我们当前的data。是不是定义一个这个呃，这个K的一个select这样一个提取器，最后提取出来的是data的user ID和data的a dad包成一个二元组返回，对吧？啊就这样啊，当然就是说大家知道如果要用下划线的话，那你是不是后面这个就没办法去去去用这个下划线了，对吧？啊，所以这里边我们把它还是写成完整的这个函数的形式。

接下来。你有一个自定义的short blacklist user。当然这里边我们可以传一个定义好的上限，对吧，超过多少次点击量的话，我们就去报警，呃，就是把它输出到这个黑名单里面，然后接下来他的操作就都没用了啊，这是我们的基本的一个想法。