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

ゝ一世哀愁。 2023-02-23 15:23 96阅读 0赞

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

**`分渠道统计`**

import java.sql.Timestamp
    import java.util.UUID
    import java.util.concurrent.TimeUnit
    
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.source.{ RichSourceFunction, SourceFunction}
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    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 MarketingUserBehavior(userId:String,behavior:String,channel: String,timestamp:Long)
    //输出结果样例类
    case class MarketingViewCount(windowStart:String,windowEnd:String,channel: String,behavior:String,count:Int)
    
    object AppMarketingByChannel { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
    
        val dataStream = env.addSource(new SimulatedEventSource())
          //下面随机生成数据 val ts =System.currentTimeMillis() 已经是毫秒了 不用*1000
          .assignAscendingTimestamps(_.timestamp)
          .filter(_.behavior != "UNISTALL")
          .map(data=>{ 
            ((data.channel,data.behavior),1L)
          })
          .keyBy(_._1)  //以渠道和行为作为key分组
          .timeWindow(Time.hours(1),Time.seconds(10))
          .process(new MarketingCountByChannel())
    
        dataStream.print()
        env.execute("app marketing by channel job")
    
    
      }
    
    }
    //自定义数据源
    class SimulatedEventSource() extends RichSourceFunction[MarketingUserBehavior]{ 
      //定义是否运行的标识位
      var running = true
      //定义用户行为的集合
      val behaviorTypes:Seq[String] = Seq("CLICK","DOWNLOAD","INSTALL","UNINSTALL")
      //定义渠道的集合
      val channelSets:Seq[String] = Seq("wechat","weibo","appstore","huaweistore")
      //定义一个随机数发生器
      val rand :Random = new Random()
      override def run(sourceContext: SourceFunction.SourceContext[MarketingUserBehavior]): Unit = { 
        //定义一个生成数据的上限
        val maxElements = Long.MaxValue
        var count = 0L
    
        //随机生成所有数据
        while (running && count < maxElements){ 
          val id = UUID.randomUUID().toString
          val behavior = behaviorTypes(rand.nextInt(behaviorTypes.size))
          val channel = channelSets(rand.nextInt(channelSets.size))
          val ts =System.currentTimeMillis()
    
          sourceContext.collect(MarketingUserBehavior(id,behavior,channel,ts))
    
          count += 1
          TimeUnit.MICROSECONDS.sleep(10L)
        }
    
      }
    
      override def cancel(): Unit = running = false
    }
    
    //自定义处理函数 第一个是keyby里面的二元组
    class MarketingCountByChannel() extends ProcessWindowFunction[((String,String),Long),MarketingViewCount,(String,
      String),TimeWindow]{ 
      override def process(key: (String, String), context: Context, elements: Iterable[((String, String), Long)],
                           out: Collector[MarketingViewCount]): Unit = { 
        //context.window.getStart是长整形 所以new 一个 变成String类型
        val startTs = new Timestamp(context.window.getStart).toString
        val endTs = new Timestamp(context.window.getEnd).toString
        val channle = key._1
        val behavior = key._2
        val count = elements.size
        out.collect(MarketingViewCount(startTs,endTs,channle,behavior,count))
      }
    }

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

import java.sql.Timestamp
    import java.util.UUID
    import java.util.concurrent.TimeUnit
    import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.source.{ RichSourceFunction, SourceFunction}
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    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 MarketingUserBehavior(userId:String,behavior:String,channel: String,timestamp:Long)
    //输出结果样例类
    case class MarketingViewCount(windowStart:String,windowEnd:String,channel: String,behavior:String,count:Long)
    
    object AppMarketing { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
    
        val dataStream = env.addSource(new SimulatedEventSource())
          //下面随机生成数据 val ts =System.currentTimeMillis() 已经是毫秒了 不用*1000
          .assignAscendingTimestamps(_.timestamp)
          .filter(_.behavior != "UNISTALL")
          .map(data => { 
            ("dummyKey", 1L)
          })
          .keyBy(_._1) //以渠道和行为作为key分组
          .timeWindow(Time.hours(1), Time.seconds(10))
          .aggregate(new CountAgg(), new MarketingCountTotal())
    
        dataStream.print()
        env.execute("app marketing job")
    
      }
    }
    //IN是输入的二元组 ((data.channel,data.behavior),1L)
    class CountAgg() extends AggregateFunction[(String,Long),Long,Long] { 
      override def createAccumulator(): Long = 0L
    
      override def add(in: (String, Long), acc: Long): Long = acc + 1
    
      override def getResult(acc: Long): Long = acc
    
      override def merge(acc: Long, acc1: Long): Long = acc + acc1
    }
    
    
    class MarketingCountTotal() extends WindowFunction[Long,MarketingViewCount,String,TimeWindow]{ 
      override def apply(key: String, window: TimeWindow, input: Iterable[Long],
                         out: Collector[MarketingViewCount]): Unit = { 
        val startTs = new Timestamp(window.getStart).toString
        val endTs = new Timestamp(window.getEnd).toString
        val count = input.iterator.next()
        out.collect(MarketingViewCount(startTs,endTs,"app marketing","total",count))
      }
    }
    
    //自定义数据源
    class SimulatedEventSource() extends RichSourceFunction[MarketingUserBehavior]{ 
      //定义是否运行的标识位
      var running = true
      //定义用户行为的集合
      val behaviorTypes:Seq[String] = Seq("CLICK","DOWNLOAD","INSTALL","UNINSTALL")
      //定义渠道的集合
      val channelSets:Seq[String] = Seq("wechat","weibo","appstore","huaweistore")
      //定义一个随机数发生器
      val rand :Random = new Random()
      override def run(sourceContext: SourceFunction.SourceContext[MarketingUserBehavior]): Unit = { 
        //定义一个生成数据的上限
        val maxElements = Long.MaxValue
        var count = 0L
    
        //随机生成所有数据
        while (running && count < maxElements){ 
          val id = UUID.randomUUID().toString
          val behavior = behaviorTypes(rand.nextInt(behaviorTypes.size))
          val channel = channelSets(rand.nextInt(channelSets.size))
          val ts =System.currentTimeMillis()
    
          sourceContext.collect(MarketingUserBehavior(id,behavior,channel,ts))
    
          count += 1
          TimeUnit.MICROSECONDS.sleep(10L)
        }
    
      }
    
      override def cancel(): Unit = running = false
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 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]: https://img-blog.csdnimg.cn/2020070722051818.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200707223234865.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1,size_16,color_FFFFFF,t_70
[20201029215409306.png_pic_center]: https://img-blog.csdnimg.cn/20201029215409306.png#pic_center