大數據挖掘算法——時間衰變算法

大家或許都聽過一個故事——「遺忘曲線」。

遺忘曲線是由德國心理學家艾賓浩斯（Hermann Ebbinghaus ）研究發現的，其描述了人類大腦對新事物遺忘的規律，人們可以從遺忘曲線中掌握遺忘規律並加以利用，從而提升自我記憶的能力。

人的記憶衰變過程，如圖1所示。

圖1  人的記憶衰變過程（來自百度百科）

這條曲線告訴人們在學習中的遺忘是有規律的，遺忘的進程很快，並且先快後慢。在分析用戶對電商平臺琳琅滿目的商品的興趣偏好的變化時，也可以借鑑遺忘曲線。

對於興趣和偏好這兩個概念，從不同的角度分析會有不同的定義，這裡從時間的角度分析。

興趣：對事物喜好或關切的情緒，它是參與實踐的基礎，可以看作是一個短期行為。隨著時間的推移，這種喜好也隨之變化。

偏好：具有濃厚的積極情緒，伴隨著成長過程中的主觀傾向，可以看作是一個長期行為，它是基本穩定的。

對於電商平臺來說，通過營銷活動達到的目的是關心用戶的「短期興趣」。看看用戶短期內會更傾向於購買哪些寶貝，從而更好地去做精準營銷，如簡訊、站內廣告等。

最簡單的場景，如果用戶在半年前購買過某件商品，但從此以後沒有再次對其產生過任何行為（瀏覽、收藏、加入購物車和購買），那麼用戶對於該商品的興趣衰變曲線如圖2所示。

圖2  單次行為的衰變曲線

重複行為是指用戶在半年內針對同一商品多次瀏覽、收藏、購買該商品。

令t1、t2、t3表示3次相鄰重複行為的時刻，用戶興趣度的衰變曲線如圖3所示。

圖3  用戶興趣度的衰變曲線

伴隨著每一次的行為調整新的初始衰變值和衰變速率。可以通過下面的函數表達式來進行更深入的理解。

符號說明：

表示用戶在j時刻以i行為作用k對象的次數。

表示時刻距離當前時間的天數。

表示衰減因子。

考慮到用戶會出現多次行為（這裡默認每次行為後的衰減因子一致），當前時刻對某商品的興趣總和公式如下。

最後結合Sigmoid函數對其進行歸一化，確定用戶對某商品的興趣值。

基於遺忘曲線的興趣度分析，它能夠從側面反映用戶對有過「行為」的所有商品進行的興趣排序，從而更好地進行商品推薦。

在分析用戶的商品推薦時，我們會選擇動手實踐其中的熵權重算法和時間衰變算法，最終結合業務的實際場景重新組合一個綜合模型。

1．數據源的獲取

這裡會考慮從HBase中讀取數據源，具體數據特徵會涉及用戶ID、商品類目、寶貝、行為類型、次數和操作時間。Apache HBase經過長達8年的發展，在2017年1月中旬又發布了新版本（Hbase 1.3.0），多個方面的性能也得到了提升。

為了能夠成功調用HBase的API，我們優先在Maven工程的pom.xml中添加如下代碼。

<dependency>             <groupId>org.apache.hbase</groupId>             <artifactId>hbase-client</artifactId>             <version>1.2.4</version></dependency>
註：這裡使用的是Hbase 1.2.4版本，可以在中央倉庫mvnrepository中搜索hbase-client來進行選擇。
接下來給出一個連接HBase的測試版本，檢測是否能夠成功獲取HBase中的表數據，代碼如下。
/**  * 掃描rowkey返回行數  *  * @param prefixRowKey rowkey前綴  * @return 行數  */
def getRowPrefixNum(table: Table,prefixRowKey: String): Option[Int] = {   var num = 1   try {       val scan: Scan = new Scan()       scan.setStartRow(Bytes.toBytes(prefixRowKey))       scan.setStopRow(Bytes.toBytes(prefixRowKey + "|"))       val resultScanner: ResultScanner = table.getScanner(scan)       val it = resultScanner.iterator()
       while (it.hasNext) {         it.next()         num += 1       }  resultScanner.close()  table.close()  Some(num)  } catch {
       case e: Exception => logger.error("統計行數出錯,{}",e.getMessage)       table.close()       Some(1)   }}

main運行的代碼模塊，可以檢測數據獲取流程是否正常，代碼如下。

def main(args: Array[String]): Unit = {       val conf = HBaseConfiguration.create()
       conf.set(HConstants.ZOOKEEPER_CLIENT_PORT, "埠號")
       conf.set(HConstants.ZOOKEEPER_QUORUM, "data1,data2,data3")       val connection= ConnectionFactory.createConnection(conf)       val table=connection.getTable(TableName.valueOf("t_user"))       val s=getRowPrefixNum(table,"rowkey")       println(s.getOrElse("0"))}
最後補充整個工程相關的依賴包，代碼如下。
import org.slf4j.{Logger, LoggerFactory}import org.apache.hadoop.hbase.{HBaseConfiguration, HConstants, TableName}import org.apache.hadoop.hbase.client._import org.apache.hadoop.hbase.util.Bytes
2．用戶行為權重的調整
這裡的數據輸入來源於從HBase獲取到的用戶數據。優先選擇用戶行為的數據計算出5種行為（瀏覽、點擊、收藏、加入購物車和購買）的權重值。
（1）確定算法過程中的統計指標，代碼如下。
val standDatas = rdd.map(_.split(SEPARATOR0)).map(record =>  {    var str = ""    for(i <- 1 until record.length) {
    //其中round為方法調用，保留4位有效數字      val standValue = round((record(i).toDouble+1)/        (indexMap.get(i).get._1+indexMap.get(i).get._2),4)
      str = str.concat((standValue*math.log(standValue)).toString). concat(SEPARATOR0)     }
     str.trim()  }).map(record =>       {       val arraySet = ArrayBuffer[Double]()
       for(i <- 0 until record.length) {           arraySet+=record(i).toDouble       }       arraySet       })

其中涉及的統計指標都會在後期的計算中用到，需要緩存在RDD中。
（2）確定指標的熵值，代碼如下。
val resultSet = ArrayBuffer[Double]()for(i <- 1 to featureNum) {
    val sumAndCount = standDatas.map(_.apply(i-1)).stats()    val value=div(sumAndCount.sum,-math.log(sumAndCount.count),4)    resultSet += value
}

這步主要是計算每個特徵向量的熵值大小，也是為計算最後的權重大小做準備的。
（3）確定特徵向量的權重值，代碼如下。
//確定每個特徵向量的權重值val weightSet = ArrayBuffer[Double]()for(i <- 0 until featureNum){
    weightSet+=div(resultSet.apply(i),resultSet.sum,4)
}
weightSet.toArray
最終將計算出的用戶行為權重單獨保存在緩存Cache中，為了後期做興趣衰變分析計算時可以再使用。
3．用戶興趣衰變的量化
結合上述的Hbase數據源和行為權重值，計算每個用戶的興趣衰變值，主要有以下兩個步驟。
（1）計算用戶興趣衰變值，代碼如下。
/* * @describe: 對興趣衰變的計算 * @param: behav為行為集,factor為衰變因子,weightSet為權重集 */  def decayAlgorithm(behav:String,factor:Double,weightSet:Map [String,Double]):Double={    val behavSet = behav.split("_")    val behavCategory = behavSet.apply(0)    val behavDiff = behavSet.apply(1).toDouble    val behavNum = behavSet.apply(2).toDouble    val interestValue = math.exp(-factor * behavDiff)*behavNum    behavCategory match {
      case "browse" => weightSet.get("browse").get*interestValue
      case "click" => weightSet.get("click").get*interestValue
      case "collect" => weightSet.get("collect").get*interestValue
      case "addCar" => weightSet.get("addCar").get*interestValue
      case "buy" => weightSet.get("buy").get*interestValue    }  }

在RDD中調用上述函數進行處理，計算用戶興趣隨著時間的衰變。
（2）採用Sigmoid進行歸一化處理，代碼如下。
/* * @describe: 對興趣衰變進行歸一化處理 * @param: decayValue為衰變的興趣度,factorsigmoid為歸一化參數 */  def decayRate(decayValue:Double,factorsigmoid:Double):Double = {    round(1.0/(1+math.exp(3.0-factorsigmoid*decayValue)),4)  }

上述是對用戶最終的興趣值進行歸一化的過程，得到用戶對寶貝列表的興趣排名，從而進行有針對性的推薦。和大家以往熟知的協同過濾推薦有所差異，基於用戶興趣偏好的衰變分析也可以做一定業務場景下的用戶推薦。
數據化運營中的精準推薦涉及的業務場景很多，更多時候會從多面分析用戶，甚至包括用戶畫像體系和商品畫像體系。