信息流曝光控量的实现和应用

val gatherid = obj.getString("gatherid")
val gatheridArr = JSON.parseArray(gatherid)
val aidArr = for(i <- (0 until gatheridArr.size()).toArray) yield{
  val objAid = gatheridArr.getJSONObject(i)
  val bucket = if(objAid.containsKey("bucket")) objAid.getString("bucket") else ""
  val vt = if(objAid.containsKey("vt")) objAid.getString("vt") else ""
  val aid = if(objAid.containsKey("aid")) objAid.getString("aid") else ""
  val key = getKey(aid,vt,bucket,obj)
  (key,(1,0))
}
aidArr.filterNot(_._1=="")
复制

//视频曝光量，映射为分类曝光量
userInfo
  .mapPartitions(it => {
    val short2category = bcShort2category.value
    it.map{case(keyAid,(exp,click)) => {
      val aid = keyAid.split("_").last
      val category = short2category.getOrElse(aid,"")
      (category,(exp,click))
    }}
  })
  .reduceByKey((a,b) => (a._1+b._1,a._2+b._2))
复制

// 每天 0 点差15s的时候redis存储过期，第二天重新开始计数
val expiresTime = ((millisOfDay - todayMillis)*0.99).toLong
val redisPrefix = "category_exp_"
partition.foreach { case (category, (exp, click)) =>
  try{
    //获取redis中存储的分类曝光量，并累加最新批次的分类曝光量，然后重新写入redis中
    val redisExp = jedis.get(redisPrefix+category)
    val newExp = if(redisExp==null) exp else exp+redisExp.toInt
    jedis.psetex(redisPrefix+category,expiresTime,newExp+"")
  }catch {
    case ex: Throwable => {
      println("categoryLimit write reids error:%s".format(ex.toString + "\n" + ex.getStackTrace.map(_.toString).mkString("\n")))
      jedis.close()
      jedis = DQRedisClient.pool.getResource
    }
  }
复制

//获取设定的分类曝光限定，和实际曝光占比做比较，计算得到过曝光和欠曝光分类
val categoryLimit = bcCategoryLimit.value
val categroyLimitExp = categoryLimit
  .map{case(category,ratio) => {
    val redisExp = jedis.get(redisPrefix+category)
    val exp = if(redisExp==null) 0 else redisExp.toInt
    (category,ratio,exp)
  }}
val allExp = categroyLimitExp.map(_._3).sum.toDouble
val categoryControl = categroyLimitExp
  .map{case(category,ratio,exp) => {
    val onlineRatio = exp/allExp
    //正差值，欠曝光；负差值，过曝光
    val controlRatio = ratio - onlineRatio
    (category,controlRatio,ratio,onlineRatio,exp)
  }}
  .sortBy(-_._2)
//计算得到欠曝光超过0.5%的分类，取3个分类
val topExpArr = categoryControl.filter(_._2>=0.005).take(3)
//如果欠曝光超过0.5%的分类补足3个，则随机取欠曝光小于0.5%的分类补足3个
val randomExpArr = categoryControl.filter(x => x._2<0.005 && x._2>0)
  .map(x => (x,rand.nextInt(30)))
  .sortBy(_._2)
  .map(x => {println("rand:%s %s".format(x._2,x._1));x._1})
  .take(3-topExpArr.size)
//生成欠曝光分类数组
val expArr = topExpArr.++(randomExpArr).map(x =>"SUBCID_%s_10".format(x._1))
//生成过曝光分类数组，取3个分类做过滤控制
val filterArr = categoryControl.filter(x => x._2 < -0.005).takeRight(3).map(x => "SUBCID_%s_10".format(x._1))
val jsonInfo = getControlJson(area,expArr,filterArr,"SUBCATEGORY")
//分类控量信息写入redis中，供引擎使用
val kvRocks = DQKVRocksClient.pool.getResource
kvRocks.setex("category_exp_control",3600,jsonInfo.toString)
复制