116.尚硅谷_Flink项目-电商用户行为分析_APP市场推广统计（一）_自定义数据源

接下来我们讲一个新的拈，就是所谓的市场营销商业指标统计分析，那首先我们主要是考虑什么，哪哪个方面呢？大家知道现在随着这个智能手机的普及，其实我们现在大多数包括电商网站啊，大多数网站大家去登录去应用产生用户行为数据的时候，其实一般都是通过这个手机登录去看的，对吧，来去浏览的，所以在这种情况下，如果说我们要考察市场营销市场推广的一个数据的话，哎，往往呢就需要去考察，比方说我们当前这个呃APP对吧，在多在哪些渠道内被用户下载，下载点击量有多少啊，然后安装量有多少对吧？呃，就是这些数据往往是我们比较关注的一个要点，那接下来我们就给大家讲一个需求，是APP市场推广的分渠道统计，那这个需求其实整体来讲也非常的简单，就是说我们就按照什么呢？先首先是收集用。

库，这个就是比方说下载安装对吧，浏览这个APP应用的这个数据收集到了之后呢，根据来源不来自于不同的这个渠道推广渠道啊，比方说你是基于这个appstore对吧，或者说我们是基于什么微博微信做的这个推广，然后用户去做了点击下载，这些我们都能收集到，那当然就可以针对这个做一个分渠道的统计了，所以大家看这也是一个统计类指标，非常典型的统计类指标，对吧？啊，所以接下来我们在代码里边先把它做一个简单的实现，那接下来我们还是在这个副项目里边啊，现在要新建一个。我们要新建一个module了，对吧，新建一个模块了。然后当前的这个模块，我们整个叫做市场营销的分析，就叫做market analysis，好，先把它创建出来，然后当前这个子模块里边我们想想需要什么特别的这个依赖吗？好像也不需要，对吧，暂时还没想到，所以我们就先放在这儿，需要的时候再说。然后我们主要还是在里边先判断这个，就是接下来我们的代码是写在这个SME下边，Java该改名改名，或者想要新建这个skyla文件夹的话，新建。

这个就是跟之前一样，然后我们会想到需要的这个数据，那是不是也应该要导入到这个source resources文件下边啊，啊，但是现在我们这个并没有，大家看之前的那个user behavior里边并没有市场推广的一些数据，对吧？啊，那个都都已经是我们登录之后用户的一些操作行为，所以这个这部分数据呢，要不就是我们自己啊，就是相当于提前的做一些假数据出来，要不那怎么办呢？有一个之前我们讲过的方式，就是可以自动的自定义一个数据源，然后自动的生成随机产生一组数据，对吧，然后流逝的不停的输入，我们去做一个测试，那这里给大家讲一个自定义数据源，然后去做测试的这个过程，好，那接下来我们主要就是在这个scla下边啊，去new一个object，还是这个带上包名，我们当前这个叫做呃，这个不是class object，然后现在我们带上包名com点。

At硅谷后边我们当前是market analysis，然后当前的这个我们就叫做呃APP市场推广的一个分渠道统计嘛，对吧，我们就叫呃这个APP market market，呃，Analysis这个不用写了啊，我们就要by channel吧，好把这个先创建出来，然后里边的整体流程其实大家知道就就差不多对吧，就是先创建环境，然后读取数据源，后边我们去呃定义，对应的就是具体的转换操作啊，我们可能要去做开窗，然后去做统计聚合，最后输出结果就完事了，主要就是这个过程。那接下来我们首先想到是不是还是得想想我们当前的这个数据长什么样啊，对吧，尽管没有一个标准的这个数据文件，但是我们首先你要做测试嘛，先想好它到底长什么样啊，那我们这里边就是定义一个输入数据的样例类。

还是case class，大家想一想，这个case class，我们这叫程，呃，Market user behavior跟之前的那个user behavior是差不多的，但是会有一点不同，对吧？那首先既然是用户的行为嘛，那是不是应该要有这个user ID对吧？呃，这个user ID肯定是要有的，我们这个呃给给一个string吧，因为后面随机生成的话，稍微简单一点啊，这个大家想给那个长整型也是一样的，然后接下来我们给一个，哎，之前我们有那个item ID，现在的这个用户行为好像就跟item没关系了，它不是在我们具体的这个应用里边，网站上面去点那个商品，对吧？哎，他直接就是比方说下载，然后或者说浏览我们的这个推广的那个APP的页面，所以这种行为的话，直接给behavior就够了，对吧，这是一个string，然后另外还需要什么呢？哎，这里可能还需要一个渠道对吧？呃，我们写的对应的这个数据大家想到。

应该也是从买点日志里边拿过来的，那你在日志里边就得去记录我当前到底是在在哪个推广渠道里边去去做这个操作，对吧？诶，所以这里边给一个这个channel，这也是一个string，最后还应该有一个time stamp长整型的一个时间戳，哎，这就是我们首先定义好的一个输入数据的样例类类型啊啊，然后接下来我们本身没有这个呃数据文件，所以说呢，我们要自定义一个可以用来测试的数据源，接下来是自定义测试数据源，好，那所以这里边我们写一个这个啊，比方说这个模拟数据源，我们就叫simulated的，呃，这个source。

然后里边，呃，这当然不需要任何的这个参数了，对吧，我们extend一个当前我们当然需要的是一个s function了啊，我直接去实现一个rich source function里边的数据类型，生成的数据类型，前面定义好的样例类market user behavior对吧？呃，然后里边我们来做一个实现，大家还记得这个就是S方式里边最起码是需要哪些具体呃，基本的这个重写的方法吗？大家可能有已经有点忘记了，对吧，之前我们说其实里边主要的方法就是一个wrong，一个cancel，大家还记得吧？哎，就是正常来讲，我们底层的那个线程会调用这里边的这个wrong方法，不停的去呃去去生成数据，那另外呢，我们还可以定义一个cancel方法，这就是表示在什么情况下啊，就把这个当前的这个生生成数据源取消掉对吧，停止生成数据，所以我们当前一般情况呢，会给一个标志位对吧？判断当前是否要继续运行啊，所以现在我们还是一样的方法啊，是否运行的标志位，大家还记得这个给一个布尔类型的叫running对吧，布尔类型的这样的一个变量啊，默认是true，然后在某种情况下就要把它制成false，哎，那什么时候制成false，调用cancel的时候制成false就完了，对吧？哎，所以这里边我们直接把这个写出来啊，比如说。

封ing就等于false，然后接下来我们还需要什么呢？诶，现在我们接下来这个用户的行为和这里边的这个渠道，大家会想到这都是要去，还有这个用户ID啊，都是要随机生成的，对不对？诶这里边如果要去做这个随机生成的话，就稍微的有一点麻烦，我们是不是得做一个预定义啊，要不然的话我怎么知道你随机生成一个string，那这个行为没有任何意义啊，我随随便生成一组这个什么，呃，Asdf对吧，乱七八糟的一个字符串，没有实际意义，如果我们想要尽量贴合实际的话，那应该把用户能够产生的行为和它这个来源的渠道都先定义好啊，那所以接下来我们就直接把这个定义在外边啊，这里边定义定义用户行为和渠道的集合，那这个集合的话大家知道。

就是你给一个list也好，或者直接给一个啊，就是我们这个序列啊，Seek也好，什么东西都好，对吧，只要把它定义在这儿就行，我们定义一个behavior set，然后或者你定义一个set也行，对吧，这个我们叫是叫成set啊，然后把它定义成一个seat好了，里边是一个string类型的集合，然后定义的时候直接写死在这里啊，比方说我们可能这个行为有哪些呢？啊，那用户可能有一个可能有一个浏览行为对吧？啊，比方说我们这个view或者呢，可能有一个就随便写啊，可能有一个点击行为，可能有一个这个下载对吧，Download。

啊，那另外还可能有这个安装之类的，当然现在可能这个安装会比较少了是吧？In store，呃，当然另外还可能有这个on ins store卸载行为，这个我们就都列在这里啊，就有可能产生的一些行为都放在，或者说你认为这个点击和这个浏览是一个行为的话，我们可以只放一个对吧？这就是按照这个业务场景的要求来自己去做设定了，然后另外我们设定一个渠道的集合channel set，那同样它也可以是一个string类型的sequence，那接下来还是直接把它写死定义在这儿对吧？呃，这里边比方说我们的来源，呃，那这个就自己随便定义了啊，有可能是这个，比方说appstore，对吧？呃，有可能，呃，这个是微博，微信微WeChat对吧？啊，那另外甚至有可能我们有一些这个。

推广渠道有可能是贴吧之类的地方啊，啊，这个就是如果我们的业务场景涉及到这些，大家都把它放进去就完事了，这是一些预定义的东西，然后另外我们可能还需要一个随机数的一个发生器，对吧？啊，这个预先我们先定义好，或者说你直接调用SC里边的那个，呃，那个random的那个单那个对象也可以，对吧？我们这里边还是先定义出来吧，定义一个random啊，它本身就是一个random，我们直接就用这个scla里边提供的这个随机数生成器就可以re，好，那接下来关键就在于这个cancel，我们也已经知道怎么做了，关键就在于wrong方法里边到底怎么样随机生成数据，然后发送出来，对吧？哎，那接下来我们要做的就是wrong方法里面做操作，这里我可以在做一个限制，之前我们没有做这个限制，就是数据不停的生成，对吧，源源不断的产生，那现在呢，我可以设置一个比方说我要。

做的这个测试，测试的数据量有一个上限，对吧，达到这个数据量我就直接停了就完事儿了，所以这里边呢，我可以定义一个，定义一个生成数据最大的数量，那这个最大的数量，比方说我这个叫max count count counts，对吧，直接就用，呃，这个大家给一个想要的数也行，或者说我这里边直接用这个当前长整型的max value也是可以的啊，这个最简单对吧？啊，就是至少暂时你做测试的时候肯定达不到嘛，那既然你要判断它是否达到了上限，那是不是还得有一个变量表示当前已经生成了多少啊对吧，你看它这个就是生成一条我就加一，生成一条加一，然后呢，看它是否达到了这个max count作为我们的一个就是是否要继续运行的一个标志啊，那这里边就是还还得定义，还得定义一个count。

一开一开始初始的时候既然长整形吗？初始的时候应该是一个零，接下来那当然就是一个while循环了，对吧，While循环。不停的产生数据，我们说是不停的产生数据，其实里边也得有停止的条件，那停止的条件一个是，哎，对，就是看这个running对吧，前面不是有一个标志位吗？如果它现在是true的话，我们继续运行，如果变成false，你说明这里边点了cancel了嘛，啊，那这里边你就直接把它停了就完了。另外还得有一个要求是什么呢？就是我们的这个当前的count数量不能超过我们定义好的那个max counts对吧？按照这个标准来，所以接下来我们就是随机生成了，呃，其实我直接写在上面就行了，随机产生数据对吧？那这里面随机一个一个来看吧，首先大家想一下这个ID怎么样随机生成呢？这前面我们定义成了它是string类型，这个就特别简单，因为可以直接用，大家看这个Java里边的UUID本身就有一个random u u ID方法对吧，或者你用我们那个随机。

输入生成器去随机生成一个长整型也是可以的啊，啊，这个无所谓，我这里边直接就用了这个UUID里边的这个方法啊，那拿到它之后呢，去做一个to string，得到我们想要的那个string类型的ID就可以了，同样接下来behavior我们要去当时定义好的这个set里边随机去取，那这个随机去取该怎么取呢？其实这个非常简单，我就直接当时我们这个behavior set对吧？Behavior set里边如果要取的话，是不是可以直接传入一个当前的那个，呃，Index那个序列第几个的那个数对吧？那这个第几个数我随机生成不就完了吗？哎，所以直接里边给的这个当前的这个啊，直接用run，这不是我们定义的随机数生成器吗？后边直接去调一个next int生成一个int，然后在哪个范围内生成呢？就用当前的behavior set。

大家想到点size是不是就可以了，对吧？这是我们随机生成这个数据的一个非常常见的方法啊，就是我预先定义好一个集合，然后呢在里边随机拿，那其实只要随机的生成一个对应的那个下标就可以了，那这个下标呢，得指定它的生成的范围对吧？呃，零到这个size之间随机去拿就完事了，这是常见的一种方式，当然另外还有就是channel也是用同样的方法可以随机生成对吧？Channel set，接下来我们去取一个用red.ne int里边给一个是channel set.set啊，所以不管你里边定义了多少预定预设的这个类型啊，都可以用这种方式去做一个实现啊。那另外我们不是还要一个时间戳吗？时间戳就更简单了，直接拿系统时间就完事，对吧？system.current ten million拿到。

就可以了，好，那接下来包装成样例类，然后输出，大家还记得这个当前这个自定义SS源里边怎么输出吗？是需要用到当前的上下文对吧？Ctx，然后ctx里边大家记得有一个collect方法，就类似于后面我们，呃，就是给大家讲到的很多那个算子啊，包括process，包括Fla map里边都有一个collector对吧？用那个collector去把数据写写入到输出缓冲区里边去，那所以这里边也是一样，我们直接调用这个Ctx.collect里边就应该是一个包装好的亮丽类类型，那里边的字段，这不就是我们定义好的吗？一个一个填进来就完事了，Channel PS对吧，直接放在这儿，然后后边这里边注意还有一个控制，因为这里边你有一个抗的判断啊，对吧，那我们得得起作用啊，起作用的话非常简单。

Count加一不就完了吗？啊，那这里边为了控制这个输出的频率，我们还是稍微的让它停一停对吧？做测试嘛，你不要那个输出的太快，所以比方说我这里边，呃，稍微这个sleep一下啊呃，这里边也不用sleep太长，比方说我就给一个50毫秒，稍微等一下，然后再生成下一个，这样的话大家就可以看到源源不断的生成数据了啊这是关于当前的这个自定义数据源的一个写法啊。