编码

import org.apache.spark.sql.SparkSession
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}

class Strawberry extends Serializable{
  
    var argsMap: Map[String,String] = _
    
    def run(args:Array[String]):Unit={
        
        //参数初始化
        arsMap = initArgs(args) //通过args(0)传参
        
        val ss = SparkSession.builder().enableHiveSupport().getOrCreate()
        
        try{
            val sparkContext=ss.sparkContext
            val outHdfs = org.apache.hadoop.fs.FileSystem.get(
                new java.net.URI("hdfs://apple"),
                new org.apache.hadoop.conf.Configuration())
            val blackWordSet:Set[String] = Set.empty[String]
            
            val blackWordFilePath=arsMap("blackWordFilePath")//比如hdfs://apple/blackWord.txt
            if(outHdfs.exists(new Path(blackWordFilePaht))){//判断hdfs文件是否存在
                //获取hdfs文件的最后修改时间
                val modifyTime=outHdfs.getFileStatus(new Path(blackWordFilePath)).getModificationTime
            }
            
            
            
            
        }finally{
            if(null != ss){
                ss.stop()
            }
        }
        
        
    }
    
    //传入的参数通过一个map统一管理
    def initArgs(args:Array[String]):Map[String,String]={
    	val argsMap=Map(
    	"appName" -> args(0),
    	"checkPoint" -> args(1),
   		"topics" -> args(2),
        "blackWordFilePath" -> args(3)
    	)
	}
    
    
}

编码优化

//对于很大的元组,可以使用case class或者 用数组定义字段

val ss = SparkSession.builder().enableHiveSupport().getOrCreate()
val sc = ss.sparkContext
val rdd1=sc.parallelize(List(("id01","jack",12,Set("piano","soccer")
                              ,Map("beijing"->"china","newyork"->"usa"))))

case class Person(id:String,name:String,age:Int
                  ,hobby:Set[String],address:Map[String,String])
//case class
val rdd2=rdd1.map(x => Person(x._1,x._2,x._3,x._4,x._5))
rdd2.map(p=>{
   p.id
   p.hobby
})

val PERSON_ARRAY=Array("id","name","age","hobby","address")
//数组按顺序定义字段,用indexof取位置,避免出现大量的x(0),x(1),难以理解
val rdd3=rdd1.map(x=>Array(x._1,x._2,x._3,x._4,x._5))
rdd3.map(x=>{
      x(PERSON_ARRAY.indexOf("id")) //等价x(0)
      x(PERSON_ARRAY.indexOf("hobby")) //等价x(3)
    })

//常用方法

//字符串 转 Long    
def stringToLongEmptyZero(ss: String): Long = {
        try {
            if (ss.isEmpty) {
                0L
            } else {
                ss.toLong
            }
        } catch {
            case e: Exception => 0L
        }
    }

// break 跳出循环

import scala.util.control.Breaks

val loop = new Breaks
loop.breakable{
    
    for(x <- aArr){
        if(flag==true){
            loop.break
        }else{
            //do something
        }    
    }
    
}

冷启动用户:冷启动召回策略

不活跃重新活跃(非冷启动用户):短时兴趣和长时兴趣;但短时兴趣已经衰减很弱

长时兴趣召回策略:
兴趣模型相对固定
更新召回列表两个动机:①用户刷新频繁,召回列表的内容很快被推送完毕
②从召回列表生成已有比较长的时间,需要重新生成召回列表来体现新增内容

热搜名称+频道+热搜热度值+来源+获取时间

平均阅读速度: 900字/分钟=15字/秒
标准时长=总字数/平均阅读速度 +图片数*5
平均阅读时长=总阅读时长/点击数

文章质量分: （1-标题党异常分+exp(-用户投诉反馈次数和) +平均阅读时长/标准时长+CTR）/4

内容标签:

通过TF-IDF生成原始关键词和关键词对应tf-idf值,
用TextRank算法计算关键词和权重,对TF-IDF值进行加权

内容多样性: 偏好分频道存,在推荐策略中推荐多个频道

########

这部分用的标签指关键词

非冷启动用户8个召回策略:

兴趣召回:

用户兴趣模型中取不超过3个兴趣标签,根据标签从内容倒排抽取30篇;

进行已推过滤,再每个标签匹配分数取top10

列表之间去重

对每个标签保留的文章进行ctr打分

兴趣扩展:

用户兴趣模型中抽取2个二级类别,根据二级类别标签从内容倒排抽取文章30篇;

进行已推过滤,再每个标签匹配分数取top10

列表之间去重

对每个标签保留的文章进行ctr打分

兴趣探索:

从用户兴趣模型之外的二级类别做探索,选定两个二级类别;

根据二级类别从内容倒排抽取文章,每个二级类别20篇;

进行已推过滤,按标签-文章分数取top10;

列表之间去重

ctr打分

兴趣协同:

根据用户所属群体,读取群体标签;

从协同标签集合中抽取至多1个协同标签;每个标签重内容倒排抽取20篇文章

进行已推过滤,按标签-文章分数取top10;

列表之间去重

ctr打分

地域召回:

根据用户请求LocalCode提取整个地域候选列表,进行已推过滤,按发布时间取top10;

列表间去重;

对文章ctr打分;

热点召回:

提取整个热门候选列表,进行已推过滤,按发布时间取top10;

列表间去重;

对文章ctr打分;

相似召回:

针对上一刷有点击/收藏/转发/分享行为的每篇文章,从相似内容列表按相似度随机抽取相似文章,各取30篇;

已推过滤,每个行为按相似度取top10;

列表之间进行去重;

对每个列表进行ctr打分;

默认召回:

兜底策略,如果其他召回无法凑齐10篇文章,则用默认召回补齐;

提取整个默认侯选池;

对默认召回 侯选池进行已推过滤后提取不超过30篇文章;

列表之间去重;

对保留文章进行ctr打分

默认侯选池是线下预先 生成的长度为1000的多样化文章集合