Flink_uv统计——使用布隆过滤器

一：UV在一定时间内(如1小时)网站的独立访客数量（Unique Visitor.）
一天内，同一访客的多次访问仅记录为一名访客。
一般通过用户IP和cookie判断UV值的两种方式。埋点日志一般包含USERID
package networkflow

import networkflow.PageView.UserBehavior
import networkflow.UniqueVistorWithBoolean.UvCount
import org.apache.flink.streaming.api._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.{AllWindowFunction, ProcessWindowFunction}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import redis.clients.jedis.Jedis

object UniqueVistorWithBoolean {
//case class UvCount(windowEnd:Long,count:Long)
case class UvCount(windowEnd: Long, count: Long)
def main(args: Array[String]): Unit = {
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)  // 转换成样例类型，提取时间戳和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)  val uvStream = dataStream   .filter(_.behavior == "pv")   .map( data => ("uv", data.userId) )   .keyBy(_._1)   .timeWindow(Time.hours(1))   .trigger(new MyTrigger())    // 定制触发器 每次数据 直接清空window   .process( new UvCountWithBloom() )  uvStream.print()  env.execute("uv with bloom job") 

}
}
// 定制触发器 每次数据 直接清空window (输入类型)，window类型】
class MyTrigger() extends Trigger[(String, Long), TimeWindow]{
override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

override def onProcessingTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

override def clear(window: TimeWindow, ctx: Trigger.TriggerContext): Unit = {}

override def onElement(element: (String, Long), timestamp: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult =
///每次数据
TriggerResult.FIRE_AND_PURGE
}

// 布隆过滤器的自定义主要是位图和hash函数
class Bloom(size: Long) extends Serializable{
private val cap = size // 默认cap应该是2的整次幂

// hash函数
def hash(value: String, seed: Int): Long = {
var result = 0
for( i <- 0 until value.length ){
result = result * seed value.charAt(i)
}
// 返回hash值，映射cap范围内 截取result
(cap - 1) & result
}
}

// 实现自定义窗口处理函数
class UvCountWithBloom() extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow]{
// 定义redis连接和布隆过滤器
lazy val jedis = new Jedis(“hadoop203”, 6379)
//64m的空间 64M=2 ^6*1000k=2610241024=2^6*220 =2^29bit 用5亿个位置保存1亿个id
lazy val bloomFilter = new Bloom(1<<29) // 位的个数：2^6(64) * 2^20(1M) * 2^3(8bit) ,64MB

// 只有在收集所有数据和窗口触发计算时才会调用；现在每个数据都被调用一次 因为用了triger
override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[UvCount]): Unit = {
// 先定义redis存储位图key 这是关窗时的时间值
val windowEnd = context.window.getEnd.toString

// 此外，当前窗口uv count值，作为状态保存redis里面，用一个叫uvcount的hash表来保存（windowEnd，count） val currentKey = context.window.getEnd.toString //当前窗口uvcount值保存到状态redis里 用一个uvcount的hash表来保存（windowEnd,count） //定义表名称 uvcount val uvCountMap = "uvcount" var count=0L if(jedis.hget(uvCountMap, currentKey) != null) {   // 从redis取出当前窗口uv count值   count=jedis.hget(uvCountMap,currentKey).toLong } // 去重:判断当前userId的hash值对应的位图位置，是否为0 val userId = elements.last._2.toString // 计算hash值对应着位图中的偏移量 61为可调参数 防止哈希碰撞 val offset = bloomFilter.hash(userId, 61) // 用redis位操作命令，取bitmap中对应位值 val isExist = jedis.getbit(windowEnd, offset) if(!isExist){   // 如果不存在，则位图对应位置1，并将count值加1   jedis.setbit(windowEnd, offset, true)   //redis存放数据（"uvcount"，context.window.getEnd.toString，count 1）   jedis.hset(uvCountMap, currentKey, (count   1).toString) } 

}
}