用戶模型簡介
知乎 AI 用戶模型服務(wù)于知乎兩億多用戶,主要為首頁、推薦、廣告、知識(shí)服務(wù)、想法、關(guān)注頁等業(yè)務(wù)場(chǎng)景提供數(shù)據(jù)和服務(wù),例如首頁個(gè)性化 Feed 的召回和排序、相關(guān)回答等用到的用戶長期興趣特征,問題路由、回答排序中用到的 TPR「作者創(chuàng)作權(quán)威度」,廣告定向投放用到的基礎(chǔ)屬性等。
主要功能
提供的數(shù)據(jù)和功能主要有:
用戶興趣:長期興趣、實(shí)時(shí)興趣、分類興趣、話題興趣、keyword 興趣、作者創(chuàng)作權(quán)威度等,
用戶 Embedding 表示:最近鄰用戶、人群劃分、特定用戶圈定等,
用戶社交屬性:用戶親密度、二度好友、共同好友、相似優(yōu)秀回答者等,
用戶實(shí)時(shí)屬性: LastN 行為、LastLogin 等,
用戶基礎(chǔ)屬性:用戶性別預(yù)測(cè)、年齡段計(jì)算、職業(yè)預(yù)估等。
服務(wù)架構(gòu)
整體主要分為 Streaming / 離線計(jì)算、在線服務(wù)和 HBase 多集群同步三部分組成,下面將依次進(jìn)行介紹。
用戶模型服務(wù)架構(gòu)圖
Streaming / 離線計(jì)算
Streaming 計(jì)算主要涉及功能 LastRead、LastSearch、LastDisplay,實(shí)時(shí)話題/ Keyword 興趣、最后登錄時(shí)間、最后活躍的省市等。
用戶模型實(shí)時(shí)興趣計(jì)算邏輯圖
實(shí)時(shí)興趣的計(jì)算流程
相應(yīng)日志獲取。從 CardshowLog、PageshowLog、QueryLog 中抽取<用戶,contentToken,actionType >等內(nèi)容。
映射到對(duì)應(yīng)的內(nèi)容維度。對(duì)于問題、回答、文章、搜索分別獲取對(duì)應(yīng)的 Topic 和 Keyword,搜索內(nèi)容對(duì)應(yīng)的 Topic。在 Redis 中用 contentToken 置換 contentId 后,請(qǐng)求 ContentProfile 獲取其對(duì)應(yīng)話題和關(guān)鍵詞;對(duì)于 Query,調(diào)用 TopicMatch 服務(wù),傳遞搜索內(nèi)容給服務(wù),服務(wù)返回其對(duì)應(yīng)的 Topic;調(diào)用 Znlp 的 KeywordExtractorJar 包,傳遞搜索內(nèi)容并獲得其對(duì)應(yīng)的 Keyword 。
用戶-內(nèi)容維度匯總。根據(jù)用戶的行為,在<用戶,topic,actionType>和<用戶,keyword,actionType>層面進(jìn)行 groupBy 聚合匯總后,并以 hashmap 的格式存儲(chǔ)到 Redis,作為計(jì)算用戶實(shí)時(shí)興趣的基礎(chǔ)數(shù)據(jù),按時(shí)間衰減系數(shù) timeDecay 進(jìn)行新舊興趣的 merge 后存儲(chǔ)。
計(jì)算興趣。在用戶的歷史基礎(chǔ)數(shù)據(jù)上,按一定的 decay 速度進(jìn)行衰減,按威爾遜置信區(qū)間計(jì)算用戶興趣 score,并以 Sortedset 的格式存儲(chǔ)到 Redis。
關(guān)于興趣計(jì)算,已經(jīng)優(yōu)化的地方主要是:如何快速的計(jì)算平滑參數(shù) alpha 和 beta,如何 daily_update 平滑參數(shù),以及用卡方計(jì)算置信度時(shí),是否加入平滑參數(shù)等都會(huì)對(duì)最終的興趣分值有很大的影響,當(dāng) display 為 1 曝光數(shù)量不足的情況下,興趣 score 和 confidence 計(jì)算出現(xiàn) 的 bias 問題等。
在線服務(wù)
隨之知乎日益增加的用戶量,以及不斷豐富的業(yè)務(wù)場(chǎng)景和與之相對(duì)應(yīng)出現(xiàn)的調(diào)用量上升等,對(duì)線上服務(wù)的穩(wěn)定性和請(qǐng)求時(shí)延要求也越來越高。 舊服務(wù)本身也存在一些問題,比如:
在線服務(wù)直連 HBase,當(dāng)數(shù)據(jù)熱點(diǎn)的時(shí)候,造成某些 Region Server 的負(fù)載很高,P95 上升,輕者造成服務(wù)抖動(dòng),監(jiān)控圖偶發(fā)有「毛刺」現(xiàn)象,重者造成服務(wù)幾分鐘的不可用,需要平臺(tái)技術(shù)人員將 Region 從負(fù)載較高的 RegionServer 上移走。
離線任務(wù)每次計(jì)算完成后一次大批量同時(shí)寫入離線和在線集群,會(huì)加重 HBase 在線集群Region Server 的負(fù)載,增大 HBase get 請(qǐng)求的時(shí)延,從而影響線上服務(wù)穩(wěn)定性和 P95。
針對(duì)問題一,我們?cè)谠瓉淼姆?wù)架構(gòu)中增加緩存機(jī)制,以此來增強(qiáng)服務(wù)的穩(wěn)定型、減小 Region Server 的負(fù)載。
針對(duì)問題二,修改了離線計(jì)算和多集群數(shù)據(jù)同步的方式,詳見「HBase多集群存儲(chǔ)機(jī)制」部分。
Cache機(jī)制具體實(shí)現(xiàn)
沒有 Cache 機(jī)制時(shí),所有的 get 和 batchGet 方法直接請(qǐng)求到 HBase,具體如下圖:
用戶模型服務(wù)請(qǐng)求序列圖
UserProfileServiceApp 啟動(dòng)服務(wù),將收到的請(qǐng)求交由 UserProfileServiceImpl 具體處理
UserProfileServiceImp 根據(jù)請(qǐng)求參數(shù),調(diào)用 GetTranslator 將 UserProfileRequest.GetRequest 轉(zhuǎn)化成 HBase 中的 Get Object(在 Map 中維護(hù)每個(gè) requestField 對(duì)應(yīng) HBase 中的 tablename,cf,column,prefix 等信息),以格式Map[String, util.List[(AvailField, Get)]]返回。
UserProfileServiceImp 用 Future 異步向 HBase 發(fā)送 get 請(qǐng)求,獲取到結(jié)果返回。
增加 Cache 機(jī)制的具體方法,在上面的第二步中,增加一個(gè) CacheMap,用來維護(hù) get 中 AvailField 對(duì)應(yīng) Cache 中的 key,key 的組成格式為:「 tablename 縮寫| columnfamily 縮寫| columnname 縮寫| rowkey 全寫」。這里使用的 Redis 數(shù)據(jù)結(jié)構(gòu)主要有兩種,SortedSet 和 Key-Value對(duì)。服務(wù)端收到請(qǐng)求后先去轉(zhuǎn)化 requestField 為 Cache 中的 key,從 Cache 中獲取數(shù)據(jù)。對(duì)于沒有獲取到 requestField 的轉(zhuǎn)化成 GetObject,請(qǐng)求 HBase 獲取,將結(jié)果保存到 Cache 中并返回。
最終效果
用戶模型的訪問量大概為 100K QPS,每個(gè)請(qǐng)求轉(zhuǎn)化為多個(gè) get 請(qǐng)求。 增加 Cache 前 get 請(qǐng)求的 P95 為30ms,增加 Cache 后降低到小于 15ms,Cache 命中率 90% 以上。
HBase 多集群存儲(chǔ)機(jī)制
離線任務(wù)和 Streaming 計(jì)算主要采用 Spark 計(jì)算實(shí)現(xiàn), 結(jié)果保存到 HBase 的幾種方式:
方法一:每次一條
1. 每次寫進(jìn)一條,調(diào)用 API 進(jìn)行存儲(chǔ)的代碼如下:
valhbaseConn=ConnectionFactory.createConnection(hbaseConf)valtable=hbaseConn.getTable(TableName.valueOf("word"))x.foreach(value=>{varput=newPut(Bytes.toBytes(value.toString))put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("c1"),Bytes.toBytes(value.toString))table.put(put)})
方法二:批量寫入
2. 批量寫入 HBase,使用的 API:
/***{@inheritDoc}*@throwsIOException*/@Overridepublicvoidput(finalList
方法三:MapReduce 的 saveAsNewAPIHadoopDataset 方式寫入
3. saveAsNewAPIHadoopDataset 是通用的保存到 Hadoop 存儲(chǔ)系統(tǒng)的方法,調(diào)用 org.apache.hadoop.mapreduce.RecordWriter 實(shí)現(xiàn)。org.apache.hadoop.hbase.mapreduce.TableOutputFormat.TableRecordWriter 是其在 HBase 中的實(shí)現(xiàn)類。底層通過調(diào)用 hbase.client.BufferedMutator.mutate() 方式保存。
valrdd=sc.makeRDD(Array(1)).flatMap(_=>0to1000000)rdd.map(x=>{varput=newPut(Bytes.toBytes(x.toString))put.addColumn(Bytes.toBytes("f1"),Bytes.toBytes("c1"),Bytes.toBytes(x.toString))(newImmutableBytesWritable,put)}).saveAsHadoopDataset(jobConf)/***Writesakey/valuepairintothetable.*@throwsIOExceptionWhenwritingfails.*/@Overridepublicvoidwrite(KEYkey,Mutationvalue)throwsIOException{if(!(valueinstanceofPut)&&!(valueinstanceofDelete)){thrownewIOException("PassaDeleteoraPut");}mutator.mutate(value);}
方法四:BulkLoad 方式
4. BulkLoad 方式,創(chuàng)建 HFiles,調(diào)用 LoadIncrementalHFiles 作業(yè)將它們移到 HBase 表中。
首先需要根據(jù)表名 getRegionLocator 得到 RegionLocator,根據(jù) RegionLocator 得到 partition,因?yàn)樵?HFile 中是有序的所以,需要調(diào)用 rdd.repartitionAndSortWithinPartitions(partitioner) 將 rdd 重新排序。
HFileOutputFormat2.configureIncrementalLoad(job,table, regionLocator) 進(jìn)行任務(wù)增量Load 到具體表的配置 實(shí)現(xiàn)并執(zhí)行映射( 并減少) 作業(yè),使用 HFileOutputFormat2 輸出格式將有序的放置或者 KeyValue 對(duì)象寫入HFile文件。Reduce階段通過調(diào)用 HFileOutputFormat2.configureIncrementalLoad 配置在場(chǎng)景后面。執(zhí)行LoadIncrementalHFiles 作業(yè)將 HFile 文件移動(dòng)到系統(tǒng)文件。
staticvoidconfigureIncrementalLoad(Jobjob,Tabletable,RegionLocatorregionLocator,Class?extends?OutputFormat,??>>cls)throwsIOException{Configurationconf=job.getConfiguration();job.setOutputKeyClass(ImmutableBytesWritable.class);job.setOutputValueClass(KeyValue.class);job.setOutputFormatClass(cls);//Basedontheconfiguredmapoutputclass,setthecorrectreducertoproperly//sorttheincomingvalues.if(KeyValue.class.equals(job.getMapOutputValueClass())){job.setReducerClass(KeyValueSortReducer.class);}elseif(Put.class.equals(job.getMapOutputValueClass())){job.setReducerClass(PutSortReducer.class);}elseif(Text.class.equals(job.getMapOutputValueClass())){job.setReducerClass(TextSortReducer.class);}else{LOG.warn("Unknownmapoutputvaluetype:"+job.getMapOutputValueClass());}conf.setStrings("io.serializations",conf.get("io.serializations"),MutationSerialization.class.getName(),ResultSerialization.class.getName(),KeyValueSerialization.class.getName());configurePartitioner(job,startKeys);//SetcompressionalgorithmsbasedoncolumnfamiliesconfigureCompression(table,conf);configureBloomType(table,conf);configureBlockSize(table,conf);configureDataBlockEncoding(table,conf);TableMapReduceUtil.addDependencyJars(job);TableMapReduceUtil.initCredentials(job);LOG.info("Incrementaltable"+table.getName()+"outputconfigured.");}publicstaticvoidconfigureIncrementalLoad(Jobjob,Tabletable,RegionLocatorregionLocator)throwsIOException{configureIncrementalLoad(job,table,regionLocator,HFileOutputFormat2.class);}valhFileLoader=newLoadIncrementalHFiles(conf)hFileLoader.doBulkLoad(hFilePath,newHTable(conf,table.getName))
將 HFile 文件 Bulk Load 到已存在的表中。 由于 HBase 的 BulkLoad 方式是繞過了 Write to WAL,Write to MemStore 及 Flush to disk 的過程,所以并不能通過 WAL 來進(jìn)行一些復(fù)制數(shù)據(jù)的操作。 由于 Bulkload 方式還是對(duì)集群 RegionServer 造成很高的負(fù)載,最終采用方案三,下面是兩個(gè)集群進(jìn)行數(shù)據(jù)同步。
存儲(chǔ)同步機(jī)制
技術(shù)選型 HBase 常見的 Replication 方法有 SnapShot、CopyTable/Export、BulkLoad、Replication、應(yīng)用層并發(fā)讀寫等。 應(yīng)用層并發(fā)讀寫 優(yōu)點(diǎn):應(yīng)用層可以自由靈活控制對(duì) HBase寫入速度,打開或關(guān)閉兩個(gè)集群間的同步,打開或關(guān)閉兩個(gè)集群間具體到表或者具體到列簇的同步,對(duì) HBase 集群性能的影響最小,缺點(diǎn)是增加了應(yīng)用層的維護(hù)成本。 初期沒有更好的集群數(shù)據(jù)同步方式的時(shí)候,用戶模型和內(nèi)容模型自己負(fù)責(zé)兩集群間的數(shù)據(jù)同步工作。
用戶模型存儲(chǔ)多機(jī)房同步架構(gòu)圖
具體實(shí)現(xiàn)細(xì)節(jié)
第一步:定義用于在 Kafka 的 Producer 和 Consumer 中流轉(zhuǎn)的統(tǒng)一數(shù)據(jù) Protobuf 格式
messageColumnValue{requiredbytesqualifier=1;......}messagePutMessage{requiredstringtablename=1;......}
第二步:發(fā)送需要同步的數(shù)據(jù)到 Kafka,(如果有必要,需要對(duì)數(shù)據(jù)做相應(yīng)的格式處理),這里對(duì)數(shù)據(jù)的處理,有兩種方式。 第一種:如果程序中有統(tǒng)一的存儲(chǔ)到 HBase 的工具(另一個(gè)項(xiàng)目是使用自定義的 HBaseHandler,業(yè)務(wù)層面只生成 tableName,rowKey,columnFamily,column 等值,由 HBaseHandler 統(tǒng)一構(gòu)建成 Put 對(duì)象,并保存 HBase 中),這種方式在業(yè)務(wù)層面改動(dòng)較小,理論上可以直接用原來的格式發(fā)給 Kafka,但是如果 HBaseHandler 處理的格式和 PutMessage 格式有不符的地方,做下適配即可。
/***tableName:hbasetablename*rdd:RDD[(rowkey,family,column,value)]*/defconvert(tableName:String,rdd:RDD):RDD={rdd.map{case(rowKey:String,family:String,column:String,value:Array[Byte])=>valmessage=KafkaMessages.newBuilder()valcolumnValue=ColumnValue.newBuilder()columnValue.set......(rowKey,message.build().toByteArray)}}
第二種:程序在 RDD 中直接構(gòu)建 HBase 的 Put 對(duì)象,調(diào)用 PairRDD 的 saveAsNewAPIHadoopDataset 方法保存到 HBase 中。此種情況,為了兼容已有的代碼,做到代碼和業(yè)務(wù)邏輯的改動(dòng)最小,發(fā)送到 Kafka 時(shí),需要將 Put 對(duì)象轉(zhuǎn)換為上面定義的 PutMessage Protobuf 格式,然后發(fā)送給 Kafka。
/***tableName:hbasetablenamne*rdd:RDD[(rowKey,put)]*/defconvert(tableName:String,familyNames:Array[String],rdd:RDD):RDD={rdd.map{case(_,put:Put)=>valmessage=PutMessage.newBuilder()for(familyName<-?familyNames){??????if(put.getFamilyMap().get(Bytes.toBytes(familyName))!=null){??????val?keyValueList?=?put.getFamilyMap()????????.asInstanceOf[java.util.ArrayList[KeyValue]].asScala????????for(?keyvalue?<-?keyValueList){??????????message.setRowkey(ByteString.copyFrom(keyvalue.getRow))????????......????????}????????message.setTablename(tableName)??????}????}????(null,?message.build().toByteArray)?}}
第三步:發(fā)送到 Kafka,不同的表發(fā)送到不同的 Topic,對(duì)每個(gè) Topic 的消費(fèi)做監(jiān)控。
/***發(fā)送rdd中的內(nèi)容到brokers的指定topic中*tableName:hbasetablenamne*rdd:RDD[(rowKey,put)]*/defsend[T](brokers:String,rdd:RDD[(String,T)],topic:String)(implicitcTag:ClassTag[T]):Unit={rdd.foreachPartition(partitionOfRecords=>{valproducer=getProducer[T](brokers)partitionOfRecords.map(r=>newProducerRecord[String,T](topic,r._1,r._2)).foreach(m=>producer.send(m))producer.close()})}
第四步:另啟動(dòng) Streaming Consumer 或者服務(wù)消費(fèi) Kafka 中內(nèi)容,將 putMessage 的 Protobuf 格式轉(zhuǎn)成 HBase 的 put 對(duì)象,同時(shí)寫入到在線 HBase 集群中。 Streaming 消費(fèi)Kafka ,不同的表發(fā)送到不同的 Topic,對(duì)每個(gè) Topic 的消費(fèi)做監(jiān)控。
valtoHBaseTagsTopic=validKafkaStreamTagsTopic.map{record=>valtableName_r=record.getTablename()valput=newPut(record.getRowkey.toByteArray)for(cv<-?record.getColumnsList)?{??????????put.addColumn(record.getFamily.toByteArray)??????????......????????}????????if(put.isEmpty){??????????(new?ImmutableBytesWritable(),?null)????????}else{??????????(new?ImmutableBytesWritable(),?put)????????}????}.filter(_._2!=null)????if(!isClean)?{??????toHbaseTagsTopic.foreachRDD?{?rdd?=>rdd.saveAsNewAPIHadoopDataset(AccessUtils.createOutputTableConfiguration(constants.Constants.NAMESPACE+":"+constants.Constants.TAGS_TOPIC_TABLE_NAME))}}
如下為另一種啟動(dòng)服務(wù)消費(fèi) Kafka 的方式。
valconsumer=newKafkaConsumer[String,Array[Byte]](probs)consumer.subscribe(topics)valrecords=consumer.poll(100)for(p<-?records.partitions)?{???val?recordsOfPartition?=?records.records(p)???recordsOfPartition.foreach?{?r?=>Try(KafkaMessages.parseFrom(r.value()))match{caseSuccess(record)=>valtableName=record.getTableNameif(validateTables.contains(tableName)){valmessageType=record.getType......try{valcolumns=record.getColumnsList.map(c=>(c.getColumn,c.getValue.toByteArray)).toArrayHBaseHandler.write(tableName)......}catch{caseex:Throwable=>LOG.error("writehbasefail")HaloClient.increment(s"content_write_hbase_fail")}}else{LOG.error(s"table$tableNameisvalid")}}}//updateoffsetvallastOffset=recordsOfPartition.get(recordsOfPartition.size-1).offset()consumer.commitSync(java.util.Collections.singletonMap(p,newOffsetAndMetadata(lastOffset+1)))}
結(jié)語
最后,目前采用的由應(yīng)用控制和管理在線離線集群的同步機(jī)制,在隨著平臺(tái)多機(jī)房項(xiàng)目的推動(dòng)下,平臺(tái)將推出 HBase 的統(tǒng)一同步機(jī)制 HRP (HBase Replication Proxy),屆時(shí)業(yè)務(wù)部門可以將更多的時(shí)間和精力集中在模型優(yōu)化層面。
-
API
+關(guān)注
關(guān)注
2文章
1499瀏覽量
61970 -
AI
+關(guān)注
關(guān)注
87文章
30763瀏覽量
268906 -
模型
+關(guān)注
關(guān)注
1文章
3229瀏覽量
48813
原文標(biāo)題:兩億多用戶,六大業(yè)務(wù)場(chǎng)景,知乎AI用戶模型服務(wù)性能如何優(yōu)化?
文章出處:【微信號(hào):rgznai100,微信公眾號(hào):rgznai100】歡迎添加關(guān)注!文章轉(zhuǎn)載請(qǐng)注明出處。
發(fā)布評(píng)論請(qǐng)先 登錄
相關(guān)推薦
評(píng)論