用戶模型簡介

知乎 AI 用戶模型服務于知乎兩億多用戶，主要為首頁、推薦、廣告、知識服務、想法、關注頁等業(yè)務場景提供數(shù)據(jù)和服務，例如首頁個性化 Feed 的召回和排序、相關回答等用到的用戶長期興趣特征，問題路由、回答排序中用到的 TPR「作者創(chuàng)作權威度」，廣告定向投放用到的基礎屬性等。

主要功能

提供的數(shù)據(jù)和功能主要有：

用戶興趣：長期興趣、實時興趣、分類興趣、話題興趣、keyword 興趣、作者創(chuàng)作權威度等，

用戶 Embedding 表示：最近鄰用戶、人群劃分、特定用戶圈定等，

用戶社交屬性：用戶親密度、二度好友、共同好友、相似優(yōu)秀回答者等，

用戶實時屬性： LastN 行為、LastLogin 等，

用戶基礎屬性：用戶性別預測、年齡段計算、職業(yè)預估等。

服務架構

整體主要分為 Streaming / 離線計算、在線服務和 HBase 多集群同步三部分組成，下面將依次進行介紹。

用戶模型服務架構圖

Streaming / 離線計算

Streaming 計算主要涉及功能 LastRead、LastSearch、LastDisplay，實時話題/ Keyword 興趣、最后登錄時間、最后活躍的省市等。

用戶模型實時興趣計算邏輯圖

實時興趣的計算流程

相應日志獲取。從 CardshowLog、PageshowLog、QueryLog 中抽取<用戶，contentToken，actionType >等內(nèi)容。

映射到對應的內(nèi)容維度。對于問題、回答、文章、搜索分別獲取對應的 Topic 和 Keyword，搜索內(nèi)容對應的 Topic。在 Redis 中用 contentToken 置換 contentId 后，請求 ContentProfile 獲取其對應話題和關鍵詞；對于 Query，調(diào)用 TopicMatch 服務，傳遞搜索內(nèi)容給服務，服務返回其對應的 Topic；調(diào)用 Znlp 的 KeywordExtractorJar 包，傳遞搜索內(nèi)容并獲得其對應的 Keyword 。

用戶-內(nèi)容維度匯總。根據(jù)用戶的行為，在<用戶，topic，actionType>和<用戶，keyword，actionType>層面進行 groupBy 聚合匯總后，并以 hashmap 的格式存儲到 Redis，作為計算用戶實時興趣的基礎數(shù)據(jù)，按時間衰減系數(shù) timeDecay 進行新舊興趣的 merge 后存儲。

計算興趣。在用戶的歷史基礎數(shù)據(jù)上，按一定的 decay 速度進行衰減，按威爾遜置信區(qū)間計算用戶興趣 score，并以 Sortedset 的格式存儲到 Redis。

關于興趣計算，已經(jīng)優(yōu)化的地方主要是：如何快速的計算平滑參數(shù) alpha 和 beta，如何 daily_update 平滑參數(shù)，以及用卡方計算置信度時，是否加入平滑參數(shù)等都會對最終的興趣分值有很大的影響，當 display 為 1 曝光數(shù)量不足的情況下，興趣 score 和 confidence 計算出現(xiàn) 的 bias 問題等。

在線服務

隨之知乎日益增加的用戶量，以及不斷豐富的業(yè)務場景和與之相對應出現(xiàn)的調(diào)用量上升等，對線上服務的穩(wěn)定性和請求時延要求也越來越高。 舊服務本身也存在一些問題，比如：

在線服務直連 HBase，當數(shù)據(jù)熱點的時候，造成某些 Region Server 的負載很高，P95 上升，輕者造成服務抖動，監(jiān)控圖偶發(fā)有「毛刺」現(xiàn)象，重者造成服務幾分鐘的不可用，需要平臺技術人員將 Region 從負載較高的 RegionServer 上移走。

離線任務每次計算完成后一次大批量同時寫入離線和在線集群，會加重 HBase 在線集群Region Server 的負載，增大 HBase get 請求的時延，從而影響線上服務穩(wěn)定性和 P95。

針對問題一，我們在原來的服務架構中增加緩存機制，以此來增強服務的穩(wěn)定型、減小 Region Server 的負載。

針對問題二，修改了離線計算和多集群數(shù)據(jù)同步的方式，詳見「HBase多集群存儲機制」部分。

Cache機制具體實現(xiàn)

沒有 Cache 機制時，所有的 get 和 batchGet 方法直接請求到 HBase，具體如下圖：

用戶模型服務請求序列圖

UserProfileServiceApp 啟動服務，將收到的請求交由 UserProfileServiceImpl 具體處理

UserProfileServiceImp 根據(jù)請求參數(shù)，調(diào)用 GetTranslator 將 UserProfileRequest.GetRequest 轉化成 HBase 中的 Get Object（在 Map 中維護每個 requestField 對應 HBase 中的 tablename，cf，column，prefix 等信息），以格式Map[String, util.List[(AvailField, Get)]]返回。

UserProfileServiceImp 用 Future 異步向 HBase 發(fā)送 get 請求，獲取到結果返回。

增加 Cache 機制的具體方法，在上面的第二步中，增加一個 CacheMap，用來維護 get 中 AvailField 對應 Cache 中的 key，key 的組成格式為：「 tablename 縮寫| columnfamily 縮寫| columnname 縮寫| rowkey 全寫」。這里使用的 Redis 數(shù)據(jù)結構主要有兩種，SortedSet 和 Key-Value對。服務端收到請求后先去轉化 requestField 為 Cache 中的 key，從 Cache 中獲取數(shù)據(jù)。對于沒有獲取到 requestField 的轉化成 GetObject，請求 HBase 獲取，將結果保存到 Cache 中并返回。

最終效果

用戶模型的訪問量大概為 100K QPS，每個請求轉化為多個 get 請求。 增加 Cache 前 get 請求的 P95 為30ms，增加 Cache 后降低到小于 15ms，Cache 命中率 90% 以上。

HBase 多集群存儲機制

離線任務和 Streaming 計算主要采用 Spark 計算實現(xiàn), 結果保存到 HBase 的幾種方式：

方法一：每次一條

1. 每次寫進一條，調(diào)用 API 進行存儲的代碼如下：

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(finalListputs)throwsIOException{getBufferedMutator().mutate(puts);if(autoFlush){flushCommits();}}

方法三：MapReduce 的 saveAsNewAPIHadoopDataset 方式寫入

3. saveAsNewAPIHadoopDataset 是通用的保存到 Hadoop 存儲系統(tǒng)的方法，調(diào)用 org.apache.hadoop.mapreduce.RecordWriter 實現(xiàn)。org.apache.hadoop.hbase.mapreduce.TableOutputFormat.TableRecordWriter 是其在 HBase 中的實現(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，因為在 HFile 中是有序的所以，需要調(diào)用 rdd.repartitionAndSortWithinPartitions(partitioner) 將 rdd 重新排序。

HFileOutputFormat2.configureIncrementalLoad(job,table, regionLocator) 進行任務增量Load 到具體表的配置 實現(xiàn)并執(zhí)行映射( 并減少) 作業(yè)，使用 HFileOutputFormat2 輸出格式將有序的放置或者 KeyValue 對象寫入HFile文件。Reduce階段通過調(diào)用 HFileOutputFormat2.configureIncrementalLoad 配置在場景后面。執(zhí)行LoadIncrementalHFiles 作業(yè)將 HFile 文件移動到系統(tǒng)文件。

staticvoidconfigureIncrementalLoad(Jobjob,Tabletable,RegionLocatorregionLocator,Class>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 來進行一些復制數(shù)據(jù)的操作。 由于 Bulkload 方式還是對集群 RegionServer 造成很高的負載，最終采用方案三，下面是兩個集群進行數(shù)據(jù)同步。

存儲同步機制

技術選型 HBase 常見的 Replication 方法有 SnapShot、CopyTable/Export、BulkLoad、Replication、應用層并發(fā)讀寫等。 應用層并發(fā)讀寫 優(yōu)點：應用層可以自由靈活控制對 HBase寫入速度，打開或關閉兩個集群間的同步，打開或關閉兩個集群間具體到表或者具體到列簇的同步，對 HBase 集群性能的影響最小，缺點是增加了應用層的維護成本。 初期沒有更好的集群數(shù)據(jù)同步方式的時候，用戶模型和內(nèi)容模型自己負責兩集群間的數(shù)據(jù)同步工作。

用戶模型存儲多機房同步架構圖

具體實現(xiàn)細節(jié)

第一步：定義用于在 Kafka 的 Producer 和 Consumer 中流轉的統(tǒng)一數(shù)據(jù) Protobuf 格式

messageColumnValue{requiredbytesqualifier=1;......}messagePutMessage{requiredstringtablename=1;......}

第二步：發(fā)送需要同步的數(shù)據(jù)到 Kafka，（如果有必要，需要對數(shù)據(jù)做相應的格式處理），這里對數(shù)據(jù)的處理，有兩種方式。 第一種：如果程序中有統(tǒng)一的存儲到 HBase 的工具（另一個項目是使用自定義的 HBaseHandler，業(yè)務層面只生成 tableName，rowKey，columnFamily，column 等值，由 HBaseHandler 統(tǒng)一構建成 Put 對象，并保存 HBase 中），這種方式在業(yè)務層面改動較小,理論上可以直接用原來的格式發(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 中直接構建 HBase 的 Put 對象，調(diào)用 PairRDD 的 saveAsNewAPIHadoopDataset 方法保存到 HBase 中。此種情況，為了兼容已有的代碼，做到代碼和業(yè)務邏輯的改動最小，發(fā)送到 Kafka 時，需要將 Put 對象轉換為上面定義的 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，對每個 Topic 的消費做監(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()})}

第四步：另啟動 Streaming Consumer 或者服務消費 Kafka 中內(nèi)容，將 putMessage 的 Protobuf 格式轉成 HBase 的 put 對象，同時寫入到在線 HBase 集群中。 Streaming 消費Kafka ，不同的表發(fā)送到不同的 Topic，對每個 Topic 的消費做監(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))}}

如下為另一種啟動服務消費 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)))}

結語

最后，目前采用的由應用控制和管理在線離線集群的同步機制，在隨著平臺多機房項目的推動下，平臺將推出 HBase 的統(tǒng)一同步機制 HRP (HBase Replication Proxy)，屆時業(yè)務部門可以將更多的時間和精力集中在模型優(yōu)化層面。