1.知识点

• scala输入输出样例类
• keyBy并行度为1计算UV的技巧

map(data => ("uv", data.userId))..keyBy(_._1)

• keyBy并行度>1 计算UV的技巧

自定义MapFunction，随机自定义key+"uv"

Random.nextString(10) + "uv"

• WindowedStream.trigger的使用
  trigger触发器，每来一条数据直接清空窗口，放到redis进行计算
• trigger返回WindowedStream,继续调用process(ProcessWindowFunction)
• WindowedStream.process()的使用
  windowStream调用接口
• 布隆过滤器的实现

2.业务目标

滚动输出最近1小时内的PV

窗口：1小时

指标：点击量

3.流程心法

总流程：创建输入输出类--->执行环境--->transform转换--->各类窗口函数的调用

主Object:

1.创建执行环境,设置时间语义，并行度等

2.transform api map转换为输入样例类,并设置watermark

3.key 定义成常量"v",那么keyBy就分为同一组，如果并行则可以自定义mapFunction

4.实现trigger

5.实现processWindowFunction

4.模块详解

4.1 创建输入输出样例类

4.2 主object实现

4.2.1 创建执行环境并添加数据源

val env = StreamExecutionEnvironment.getExecutionEnvironment    
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)    
  env.setParallelism(1)     
// 从文件中读取数据    
val resource = getClass.getResource("/UserBehavior.csv")    
val inputStream: DataStream[String] = env.readTextFile(resource.getPath)

4.2.2 Datastream map转换为输入样例类

 // 转换成样例类类型并提取时间戳和watermark    
val dataStream: DataStream[UserBehavior] = inputStream      
  .map(data => {        
    val arr = data.split(",")        
    UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, arr(3), arr(4).toLong)      
  })      
  .assignAscendingTimestamps(_.timestamp * 1000L)

4.2.3 处理逻辑（1）----filter类型，timeWindow

val uvStream = dataStream      
    .filter(_.behavior == "pv")      
    .map( data => ("uv", data.userId) ) //如果要并行，并行自定义mymapper      
    .keyBy(_._1)      
    .timeWindow(Time.hours(1))  //滚动窗口      
    .trigger(new MyTrigger())  //trigger触发器，每来一条数据直接清空窗口，放到redis计算。      
    .process( new UvCountWithBloom() )

4.2.4 处理逻辑（2）----Trigger实现

class MyTrigger() extends Trigger[(String,Long),TimeWindow]{  
  override def onElement(t: (String, Long), l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = {     
    TriggerResult.FIRE_AND_PURGE   
  }   
  //系统时间有进展时做什么操作  
  override def onProcessingTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE   

  //watermark改变做什么操作  
  override def onEventTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE   

  override def clear(w: TimeWindow, triggerContext: Trigger.TriggerContext): Unit = {   
  }
}

4.2.5 处理逻辑（2）----ProcessWindowFunction实现

1.定义redis中存储位图的key ，本例为窗口结束时间

2.定义一个redis hash表，保存统计之后的每个窗口结束时间的uv count.

表名：uvcount

KEY: 窗口结束时间

VALUE:uv count值

3. 对userid进行hash, 从位图中查看hash后的偏移量是否窜在，若存在则uvcount不操作。若不存在则uvcount+1,位图也相应更新

class UvCountWithBloom() extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow]{  
  // 定义redis连接以及布隆过滤器  
  lazy val jedis = new Jedis("localhost", 6379)  
  lazy val bloomFilter = new Bloom(1<<29)    // 2的29次方，1左移29位。 位的个数：2^6(64) * 2^20(1M) * 2^3(8bit) ,64MB   

  // 本来是收集齐所有数据、窗口触发计算的时候才会调用；现在每来一条数据都调用一次  
  override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[UvCount]): Unit = {    
    // 先定义redis中存储位图的key    
    val storedBitMapKey = context.window.getEnd.toString      
    
    //另外将当前窗口的uv count值，作为状态保存到redis里，用一个叫做uvcount的hash表来保存（windowEnd，count）    
    val uvCountMap = "uvcount"    
    val currentKey = context.window.getEnd.toString    
    var count = 0L     

    // 从redis中取出当前窗口的uv count值    
    if(jedis.hget(uvCountMap, currentKey) != null)      
      count = jedis.hget(uvCountMap, currentKey).toLong     
    
    // 去重：判断当前userId的hash值对应的位图位置，是否为0    
    val userId = elements.last._2.toString    
    // 计算hash值，就对应着位图中的偏移量    
    val offset = bloomFilter.hash(userId, 61)    
    val isExist = jedis.getbit(storedBitMapKey, offset)     

    if(!isExist){      
      // 如果不存在，那么位图对应位置置1，并且将count值加1            
      jedis.setbit(storedBitMapKey, offset, true)      
      jedis.hset(uvCountMap, currentKey, (count + 1).toString)    
    }  
  }
}

4.2.6 处理逻辑（3）----布隆过滤器实现

也可以调用外部google等现成的布隆过滤器.

设计布隆过滤器的要点：
1.选好点的hash函数
2.不同userid经过hash到同一位上。不要那么稠密。
即1亿的user，我们给出2亿的位，出现碰撞的概率就特别小。
10B * 1亿，大概1GB, 用位来存，1bit * 1亿 大概10m，放redis放内存都是个很好的 选择。
即使我们扩大位防止碰撞，放6亿，也是68M，可以放到redis中。有可能出现hash碰撞

class Bloom(size: Long) extends Serializable{  
  private val cap = size    // 默认cap应该是2的整次幂   

  //hash函数 value即userid,seed随机数种子  
  def hash(value: String, seed: Int): Long = {    
    var result = 0    
    //遍历userid，对每一位进行随机数种子的处理    
    for( i <- 0 until value.length ){      
      result = result * seed + value.charAt(i)    
    }     
    // 返回hash值，要映射到cap范围内    
    (cap - 1) & result  
  }
}

4.3 完整代码