022_尚硅谷大数据技术_用户行为数据分析Flink项目_市场推广渠道统计（一）_整体架构和自定义测试数据源

来我们开始新的一个模块的分析，这个模块主要分析的是市场营销相关的一些商业指标统计，然后在这个模块里边呢，我们主要想要做的一个是APP市场推广统计，另外一个是页面的广告分析，首先来看这个APP市场推广统计，那大家知道现在随着这个智能手机普及啊，那绝大多数像这个电商应用，或者说其他的一些应用，我们的入口一般都不是传统的这个电脑网页了，呃，就是以前的传统的话，大家可能想到我一个网站啊，一个一个这样的一个应用，可能入口都得是你打开一个网页，然后去登录，然后去做一些操作，那现在的话大家手机太方便了，所以一般情况我们都是要靠APP去，呃，去这个获取新的客户流量，对吧，用户啊，那所以现在我们做推广的时候，往往更多的也是要看当前的这个APP端，APP这边到底有多少人去下载，有多少人去做新的。

这个注册，呃，所以呢，做这个市场分析的时候，我们往往就会考察不同的推广渠道，对于我们当前这个呃APP市场营销的这个推广程度啊，起到的贡献，起到的作用到底有多大啊，那大家知道，一般情况我们可能呃，你做这个推广的时候，有这个像我们现在啊，有这个公众号对吧，有这个呃微信的小程序啊对吧？有其他的一些推广渠道，比方说还可以有这个微博上推广啊，对吧？啊或者说这个微信有一些这个朋友圈，大家也看到也有广告对吧？呃，各种各样不同的推广渠道，其实都是可以收集它对应的那个数据的，比方说这就是我们说的那个导流的问题，对吧，所谓的一个公众号，如果什么呃，经常是这个10万加的这个浏览量的话，如果他给我们去做一个推广，那有可能直接这个点击量下载量就非常大，对吧，那我们可以就做一个做一个分析，做一个统计看看哦，这边这个平平台这个渠道。

推广的力度特别的给力啊，一下子做了一个广告，一下子这个就下载了很多啊，注册了很多新用户，那我们就想到这个渠道是不是可以加大投入力度啊啊，这对于公司的这个市场营销策略调整来讲是非常有帮助的，所以接下来呢，我们就来考虑这样一个场景来统计当前的市场，APP市场推广分渠道做一个统计，对吧，不同的渠道到底有哪有多少点击量，下载量。那现在我们还是在代码里边直接新建一个模块。当前的这个模块，呃，New一个module，当前这个模块我们直接就叫做呃，市场分析对吧？呃，当前这个我就叫做market analysis。然后把这个创建出来之后，里边首先应该是，呃，大家首先想一下，这个po文件里边应该也不需要什么特别的是吧，至少目前还想不到啊，呃，只要有这个flink卡夫卡相关的依赖也就够了，那这个我就先不需要了，接下来我们考虑到里边是要把这个。

当前的这个代码写到sourcew Java下边，然后当前这个resources下边呢，需要放的是这个，呃，我们对应的这个数据文件啊，因为我们现在不是从卡夫卡里边直接直接读取数据的，但是现在大家考虑我们如果想统计这个APP的推广渠道的话，在之前我们那user behavior里边是不是好像没有这个数据啊，没有对应的啊，哪个渠道对吧，用户这个点击了多少，呃，下载了多少，那所以这就我们只好自己去生成这个测试数据了，那生成测试数据两种方式，一种就是我们用脚本生成，直接写在文件里边，还有另外一种方式，因为大家想这个写在文件里的话，读取文件这感觉又不像一个真实的流处理的环境，如果要是模拟真实的流处理环境的话啊，当然我们可以说就是你随机生成数据之后换到卡夫卡里面去，对吧，然后再消消费卡夫卡，那另外还有一种更简单的形式，其实是。

我们是不是可以直接自定义一个测试数据源啊，对吧，自定义一个source function，然后直接去产出数据就可以了啊，那所以在这个之前，我们还是首先考察一下当前这个数据里边。是不是应该要有一些对应的数据类型啊，你要产生的这个测试数据源，我们还是想把它包装成一个po类嘛，那到底长什么样呢？诶，我们先考虑一下，我先新建一个package啊，带上包名com.at。硅谷点当前是market analysis，呃，当前是这个beans，对吧，我先把它放在这儿，接下来新建一个Java class，呃，首先我想到现在的这个应该还是一个用户行为对吧？那就是如果说我在这个呃，某一些平台上用户做了一个点击，或者说做了一个这个对应的呃下载对吧，或者做了一个安装这样的一个行为的话，那我们这里边应该是会收到他的一条这样的一个日志记录的，那这样的一个数据，我就把它包装成叫做marketing。

User behavior其实还是一个用户行为嘛，大家想一下，这个数据里边我主要得有哪些字段呢？哎，大家考虑考虑到这个，那应该主要就还是哪个用户，然后做了一个什么操作对不对，哎，主要就是这些啊呃，那这里边首先用户ID，那我们还是用这个长整型吧，对吧，User ID先定义出来，然后另外诶之前我们是还有那个item ID，现在的话，这其实用户对我们整个平台APPAPP的一个操作，那其实不需要什么item ID对吧？诶这里边只要有。对应的一个行为就可以了，比方说我们定义一个behavior。

然后另外大家想到behavior啊，加一个A，然后大家想到是不是还应该有一个对应的渠道名称啊，对吧，你到底是从哪个渠道做了这样一个行为，所以接下来我再定义一个string类型的channel，哦，好像我们关心的就这些数据了，最后还应该有一个时间戳，对吧？当前你这个数据发生的时间到底是什么时候？同样我们还是定义一个长整型time stepmp，这就是我们想到的啊需要的字段，呃，生成的数据我就直接包装成这样一个破类是不是就可以了？啊，那接下来我们把这个对应的构造方法还是先创建出来空参的构造方法。然后带参数的啊，也创建出来，下边还有对应的get set的这些这些方法全部生成啊，另外还是方便大家这个看到它打印输出的结果的话，我们把那个托斯逊直接生成出来啊，这就是我们的一些准备工作啊，先把它做这样的一个处理，然后接下来大家会想到，那那最后我得到的呃，对应的这个结果，那应该是一个什么样的信息呢？

最后得到的结果，我们想统计的其实就应该是某一个，诶，某一个渠道，然后比方说我们点击，然后点击了多少多少次，大家想是不是又是一个count类似的包装啊，诶，那对应的我们前面不是还要把那个窗口信息也包装上吗？因为我们统计这个是不是也应该要开窗统计啊，啊，你不可能说是这个所有全量数据，对吧，我肯定是统计一段时间内到底这个下载量，点击量对吧，到底有多少？呃，所以接下来我可能包装的又是，呃，有一个当前的这个推广渠道，然后呢，另外大家想到，如果说我这个渠道里边有点击，有下载，有安装，那这其实是不同的行为，对吧，不同行为，呃，我们可能想到它的权重是不一样的，那我可能应该要分别统计，所以我的统计应该是。

每一个渠道，然后每一种行为对应在某一个时间时间段，也就是某一个时间窗口内被操作了多少次，对不对，统计它的那个count值啊，所以接下来我给大家包装一个又是类似于之前我们那个item view count一样的东西，对吧？啊，我把这个叫做。这个channel，呃，推广操作啊，我叫做这个promotion吧，啊，大家大概知道是什么意思就行，对吧，推市场的这个推广的一个技术count，把它叫做channel promotion count，然后这里面我们关心的字段就是private，首先是不是应该得有一个channel名称啊。Channel，然后另外private应该要有一个当前的行为的名称behavior。Behavior，哦，另外我们还关心一个当前的窗口的信息，对吧？呃，长整型的那个，呃，窗口的window end，那或者这里边呢，我甚至还可以定义成string，就是说最后我不想只看那个长整型的那个时间戳，我想把它转换成一个年月日十分秒，这个可以不可以，哎，也可以，对吧？没问题啊，我们转换一下就行了嘛，我把这个叫window end，对吧？然后最后还有一个count对不对，这个是private成整型。

好，我们还是对应的啊，空参的构造器先写出来，然后带参数的写出来，下边get set对应的也要。全部写出来生成。哦，那对应的图th string，这个本来没有强制的需要强制的要求啊，但是我们一般为了方便大家打印输出，看起来舒服一点，还是把它直接生成好，那接下来这个就都搞定了，对吧？啊，那接下来我们还是在Java下边就要创建一个我们真正处理的那个class了，那当前我们这个处理的class呢？呃，大家可以想到我当前其实是带上包名啊，com.at，硅谷点market analysis哦，这里边我其实想要的就是当前APP的一个市场推广的统计，对吧，我直接就叫做APP，呃，Marketing。

By channel吧，按照不同的渠道做的一个市场推广的统计，对吧？好，把这个先列出来。那里边首先PSVM这里还是要extend，呃，这里throw一个exception。那下边我还是先把当前的流失执行环境先获取到get execution environment，对吧，定义成env，然后下边我们还是不是正确性，先把这个并行度设成全局设成一，下边设置时间语义，大家想我们现在既然数据里边带着时间戳对吧，那一般情况我们都就用这个实验时间了，对不对啊，它准确性更高一点啊，所以直接用even time，好，这是第一步。

然后接下来第二步。啊，我们把这个接下来这个叫第一步吧，对吧，因为前面这个是基本的环境准备啊，然后这里面我们是从自定义数据源中读取数据，对吧？啊，那其实大家想到这个，我们最终得到的应该就是一个data stream。是不是我们包装好的那个po类型啊，对吧，我们那个叫marketing。User behavior就包装成这样的一个数据类型啊，那当然这个如果要读的话，是不是应该是env。直接，因为我们是自定义测试数据源嘛，我是不是直接at source啊，然后要去。

自己创建一个source function啊，那所以这里边我把这个名称单独定义出来吧，比方说我把这个叫做模拟的，呃，市场的用户行为啊，所以这个我叫做simulated，大家知道这个是模拟的意思，Simulated marketing marketing。User behavior。呃，大家如果写的全面一点的话，可以把它叫S，对吧，我可以写长一点啊，大家知道具体什么含义就可以了，好把这个先定义出来，下边就是要实现了，实现自定义的模拟，呃，这个市场用户行为。数据源public static class，好，把这个写在这儿，我们接下来要实现的接口应该是implement，一个source function，对吧？

S方式啊，当然如果说你用到了更多的，比方说生命周期对吧，相关的一些东西还想定义状态的话，是不是我可以定义那个rich s function啊，啊，对吧，这个可以做的事情还是很多的啊，然后source function，大家看到它里边有一个类型泛型，当然就是当前生成的对应的那个数据类型，对吧？我们当前要生成的就是marketing user behavior。大家看把这个一写上面类型就对了，对吧？啊，这个不报错了啊，然后里边大家知道必须要实验的方法是不是有两个啊，一个叫做wrong，一个叫做cancel，这个大家知道wrong表示的其实就是只要我当前这个任务执行起来，S任务执行起来，这个wrong是不是就相当于在一直在运行啊，啊那么正常情况下，它里边这个代码是在不停的执行的，然后那个cancel呢，是不是就要控制这里边什么时候取消，什么时候就不要发送数据了，那家想我这里面怎么控制它，对外面是不是需要有一个标志位啊，对吧？哎，我们这里边。

控制，控制是否正常运行的标志位。呃，那这里边我们直接就定义一个布尔类型booing对吧，Running默认的时候应该是一个处对吧，因为它默认是不是一直在运行啊啊，那然后如果要cancel的时候怎么办。这非常简单，是不是直接running把它赋值，赋一个false是不是就搞定了？哎，这个非常简单啊，那接下来关键在于这个，呃，这个就是run里边我们要做什么事情，那大家想现在我们run的时候也没什么具体的要求，那我是不是就是直接随机生成这样的这个market user behavior就可以了，对吧？不停的随机生成对应的这个数据就完事了，那这里边随机生成的话，你比方说这个user ID user ID这个我还好想一点啊，就直接呃，Random直接去生成，对吧？那你假如说这个后面要这个behavior和这个对应的这个channel的话，难道是说我随机生成一个string吗？那就那就太假了点是吧，我们要模拟，至少你也看的像一点嘛，所以模拟生成数据的时候，我是不是应该先定义一个对，就是相当于用户的行为和渠道可以选择的范围，对不对啊，先把这个集合先定义出来，在这里啊，我们。

定义。用户行为和渠道的范围啊，那所以这里边我们其实就是一个集合了啊，比方说我定义成一个a list吧，所以这里边A点我可以直接用这个as list，然后把里边的呃，所有的那个字段是不是都直接传进去啊，对吧？所以这里边比方说我给一个呃，比方说给一个当前的这个行为啊click对吧。

直接写一下这个，然后呃，比方说还有哪些行为，大家可以想一下，比方说有一个下载对吧，Download。Download，另外还有安装对吧？In install啊，当然了还可以有卸载对不对，这也算是一个行为吧，就只要我们能统计到啊，啊，但是这个这个统计好像不应该是在渠道里边统计的是吧？啊，我只是给大家列出来啊啊这样的话我们就可以看到这些东西，然后呃，我们把这个做一个诶。哦，大家看到前面这里边这个，呃，定义的这个at least这里边没法调用是吧，这个AR我们看一下。Java u AR，我们前面可以把这个直接定义出来啊，List，当前我这个应该是一个string，对不对，把这个引入啊。然后接下来我把这个叫做当前的行为behavior list，然后等于a raise at least，对吧？啊，这样的话我知道它对应的类型是什么，这个就可以推断出来了，好，呃，那这是当前的行为的集合，然后另外是不是还有渠道的集合啊，啊，这个类似的啊，也是一个list string。

里边我定义一个当前这个channel channel list，同样ar.as list，我把对应的这个行为，呃，对应的渠道做一个列举，比方说我们现在的渠道，呃，这个appstore，对吧。这这个就随便写了啊，这个不一定符合我们真正的生产实际啊，比方说我们有这个微信对吧？啊，这个微信就不一定是公众号还是什么了啊，有可能我们是不同的公众号都分开对吧，在这里边都直接做一个统计，因为我们对应的那个推广肯定是有固定的地方的嘛，比方说还有这个微博对吧？啊，就是随便可以有这样的一些推广渠道，然后接下来我们就在里边要随机生成对应的数据了啊，那我们还是在外边定义一个随机数生成器吧啊，外面我们定义一个。

Random对吧，你有一个random。诶，那那这里边因为是在外边，就是我这里边需要去把它直接写出来啊random，你有一个random对吧。好，这样的话我就知道了，当前用这个随机数生成器，在这每一个对应的这个集合里边，列表里边去随机提取一个对应的字段，然后生成是不是就可以了啊，这其实整体流程就是这样啊，所以接下来我就是直接一个well，当前是不是running啊，按照running来做一个控制无限循环，随机生成数据啊，随机生成所有字段，那首先是那个ID user ID大家知道它应该是一个长整型，对吧，直接用random是不是next long就可以了，对吧，直接生成就完事啊，然后接下来是一个string类型的behavior，这个要生成的时候，那我们是不是应该。

诶，大家想我接下来用random要去随机生成的是不是一个一个int啊，然后是不是应该在当前的这个长度范围内的一个int呀，然后我再从这个里边去取就完事了，所以我用一个behavior list，然后是不是直接get里边这个get的当前的index应该随机生成，哎，所以random接下来是不是next in next int里边大家看是不是可以传一个范围啊，传一个棒的对吧？这样的话就是生成零到当前这个范围的一个数啊，那所以我这里边直接就传当前的behavior list.size是不是就搞定了啊，所以这个这种方法还经常用的对吧？后边同样的来一个channel，这个也是换汤换汤不换药啊，都一样，Get里边random.next int里边传一个channel list.size对吧？

在这个范围内传就可以了啊，最后还有一个时间戳，那时间戳的话，我们是不是直接用system，直接按照当前时间生成就行了，这个不要随机对吧？呃，直接system current time minutes，这是一个。长整星毫秒数对吧，那接下来就是要发出数据，大家还记得怎么样发出数据吗？当前要发出数据是不是全靠当前这个ctx上下文啊，哎，所以当前ctx里边有一个方法叫做collect，对，就像我们那个al.collect一样啊，这里边直接就可以把当前的数据收集起来，发送出去，我们想要的是一个。Marketing user behavior里边的字段，呃，一个一个写进来对吧？Behavior，然后channel，另外还有time step，哎，这就是我们做的这个操作对吧？当然也可以控制一下这个输出的频率，不要太快的话，那我做一个thread sleep一下，对吧？比方说这个我隔100毫秒生成一个数吧，对吧？这频率稍微的低一点啊，这就是我们做的一个这个测试数据源啊，我们先把这个放在这里。