Flink-电商用户行为分析(APP市场推广渠道统计)_分渠道统计+不分渠道统计_2

ゝ一纸荒年。 2022-11-21 03:47 170阅读 0赞

随着智能手机的普及，在如今的电商网站中已经有越来越多的用户来自移动端，相比起传统浏览器的登录方式，手机 APP 成为了更多用户访问电商网站的首选。对于电商企业来说，一般会通过各种不同的渠道对自己的 APP 进行市场推广，而这些渠道的统计数据（比如，不同网站上广告链接的点击量、APP 下载量）就成了市场营销的重要商业指标。 首 先 我 们 考 察 分 渠 道 的 市 场 推 广 统 计 。由于没有现成的数据，所以我们需要\*\*`自定义一个测试源来生成用户行为的事件流`\*\*。

**`分渠道统计`**

import java.sql.Timestamp
    import java.util.UUID
    
    
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.source.{ RichParallelSourceFunction, SourceFunction}
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    import scala.util.Random
    
    //定义输入数据样例类
    case class MarketUserBehavior(UserId:String,behavior:String,channle:String,timestamp:Long)
    //定义输出统计的样例类
    case class MarketCount(windowStatrt:String,windowEnd:String,channe:String,behavior:String,count:Long)
    
    
    
    object AppMarketingByChannel { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        val dataStream = env.addSource(new SimulateMarketEventSource())
          .assignAscendingTimestamps(_.timestamp)
    
        val resultStream = dataStream
          .filter(_.behavior != "Uninstall") //过滤卸载行为
          .keyBy(data => (data.channle, data.behavior)) //按照渠道和行为类型分组
          .timeWindow(Time.hours(1), Time.seconds(5))
          .process(new MarketingCountByChannel())
    
        resultStream.print()
    
        env.execute("market count by channel job")
      }
    
    }
    
    
    //自定义ProcessWindowFunction
    class  MarketCountByChannel() extends ProcessWindowFunction[MarketUserBehavior,MarketCount,(String,String),TimeWindow]{ 
      override def process(key: (String, String), context: Context, elements: Iterable[MarketUserBehavior],
                           out: Collector[MarketCount]): Unit = { 
        val windowStart = new Timestamp(context.window.getStart).toString
        val windowEnd = new Timestamp(context.window.getEnd).toString
        val channel = key._1
        val behavior = key._2
        val count = elements.size
        out.collect(MarketCount(windowStart,windowEnd,channel,behavior,count))
      }
    }
    
    
    //自定义测输数据源
    class SimulateMarketEventSource() extends RichParallelSourceFunction[MarketUserBehavior]{ 
      //定义是否在运行的标志位
      var running : Boolean = true
      //定义用户行为和推广渠道的集合
      val behaviorSet:Seq[String] = Seq("Click","Downlaod","Install","Uninstall")
      val channelSet:Seq[String] = Seq("appstore","huaweistore","weibo","weichat")
      //定义随机数生成器
      val rand:Random = Random
    
      override def cancel(): Unit = running = false
    
      override def run(ctx: SourceFunction.SourceContext[MarketUserBehavior]): Unit = { 
        //定义一个发出数据的最大量，用于控制测试数据量
        val maxCounts = Long.MaxValue
        var count = 0L
    
        //while循环，不停的随机生成数据
        while (running && count<maxCounts){ 
          val id = UUID.randomUUID().toString
          val behavior = behaviorSet(rand.nextInt(behaviorSet.size))
          val channel = channelSet(rand.nextInt(channelSet.size))
          val ts = System.currentTimeMillis()
    
          ctx.collect(MarketUserBehavior(id,behavior,channel,ts))
    
          count += 1
    
          Thread.sleep(50L)
    
        }
      }
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]  
**`不分渠道统计`**

import java.util.UUID
    import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.source.{ RichParallelSourceFunction, SourceFunction}
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.scala.function.WindowFunction
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    import scala.util.Random
    
    //定义输入数据样例类
    case class MarketUserBehavior(UserId:String,behavior:String,channle:String,timestamp:Long)
    //定义输出统计的样例类
    case class MarketCount(windowStatrt:String,windowEnd:String,channe:String,behavior:String,count:Long)
    
    
    
    object AppMarketingTotal { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        val dataStream = env.addSource(new SimulateMarketEventSource())
          .assignAscendingTimestamps(_.timestamp)
    
        val resultStream = dataStream
          .filter(_.behavior != "Uninstall") //过滤卸载行为
          .map(data => ("total",1L))
          .keyBy(_._1) //按照渠道和行为类型分组
          .timeWindow(Time.hours(1), Time.seconds(5))
            .aggregate(new MarketCountAgg(),new MarketCountResult())
    
        resultStream.print()
    
        env.execute("market count by channel job")
      }
    
    }
    
    
    //自定义预聚合函数
    class MarketCountAgg() extends AggregateFunction[(String,Long),Long,Long]{ 
      override def createAccumulator(): Long = 0L
    
      override def add(value: (String, Long), accumulator: Long): Long = accumulator + 1
    
      override def getResult(accumulator: Long): Long = accumulator
    
      override def merge(a: Long, b: Long): Long = a + b
    }
    
    //自定义窗口函数 第三个参数是keyby中的类型
    class MarketCountResult() extends WindowFunction[Long,MarketCount,String,TimeWindow]{ 
      override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[MarketCount]): Unit = { 
        out.collect(MarketCount(window.getStart.toString,window.getEnd.toString,"total","total",input.head))
      }
    }
    
    //自定义数据源
    class SimulateMarketEventSource() extends RichParallelSourceFunction[MarketUserBehavior]{ 
      //定义是否在运行的标志位
      var running : Boolean = true
      //定义用户行为和推广渠道的集合
      val behaviorSet:Seq[String] = Seq("Click","Downlaod","Install","Uninstall")
      val channelSet:Seq[String] = Seq("appstore","huaweistore","weibo","weichat")
      //定义随机数生成器
      val rand:Random = Random
    
      override def cancel(): Unit = running = false
    
      override def run(ctx: SourceFunction.SourceContext[MarketUserBehavior]): Unit = { 
        //定义一个发出数据的最大量，用于控制测试数据量
        val maxCounts = Long.MaxValue
        var count = 0L
    
        //while循环，不停的随机生成数据
        while (running && count<maxCounts){ 
          val id = UUID.randomUUID().toString
          val behavior = behaviorSet(rand.nextInt(behaviorSet.size))
          val channel = channelSet(rand.nextInt(channelSet.size))
          val ts = System.currentTimeMillis()
    
          ctx.collect(MarketUserBehavior(id,behavior,channel,ts))
    
          count += 1
          
          //休息50毫秒生成1个数
          Thread.sleep(50L)
    
        }
      }
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]

## 总结 ##

对于继承AggregateFunction中的第一个参数就是map，如果没有就是输出样例类！  
如果是叫输出key，那么就是KeyBy的类型  
![在这里插入图片描述][20201029215409306.png_pic_center]  
**`对于aggregate和使用process中继承ProcessWindowFunction是一样的`**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/12c44456f2344e24a7119d44082574d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/9ab83077e4e843e89f5c50bc64d9af00.png
[20201029215409306.png_pic_center]: https://img-blog.csdnimg.cn/20201029215409306.png#pic_center