使用者行為分析模型實踐（一）—— 路徑分析模型
一、需求背景
在網際網路資料化運營實踐中，有一類資料分析應用是網際網路行業所獨有的——路徑分析。路徑分析應用是對特定頁面的上下游進行視覺化展示並分析使用者在使用產品時的路徑分佈情況。比如：當用戶使用某APP時，是怎樣從【首頁】進入【詳情頁】的，使用者從【首頁】分別進入【詳情頁】、【播放頁】、【下載頁】的比例是怎樣的，以及可以幫助我們分析使用者離開的節點是什麼。
在場景對應到具體的技術方案設計上，我們將訪問資料根據session劃分，挖掘出使用者頻繁訪問的路徑；功能上允許使用者即時檢視所選節點相關路徑，支援使用者自定義設定路徑的起點或終點，並支援按照業務新增使用者/活躍使用者檢視不同目標人群在同一條行為路徑上的轉化結果分析，滿足精細化分析的需求。
1.1 應用場景
通常使用者在需要進行路徑分析的場景時關注的主要問題：
- 按轉換率從高至低排列在APP內使用者的主要路徑是什麼；
- 使用者在離開預想的路徑後，實際走向是什麼？
- 不同特徵的使用者行為路徑有什麼差異？
通過一個實際的業務場景我們可以看下路徑分析模型是如何解決此類問題的；
【業務場景】
分析“活躍使用者”到達目標落地頁[小視訊頁]的主要行為路徑（日資料量為十億級，要求計算結果產出時間1s左右）
【使用者操作】
- 選擇起始/結束頁面，新增篩選條件“使用者”；
- 選擇型別“訪問次數”/“會話次數”；
- 點選查詢，即時產出結果。
二、基本概念
在進行具體的資料模型和工程架構設計前，先介紹一些基礎概念，幫助大家更好的理解本文。
2.1 路徑分析
路徑分析是常用的資料挖據方法之一， 主要用於分析使用者在使用產品時的路徑分佈情況，挖掘出使用者的頻繁訪問路徑。與漏斗功能一樣，路徑分析會探索使用者在您的網站或應用上逗留的過程中採取的各項步驟，但路徑分析可隨機對多條路徑進行研究，而不僅僅是分析一條預先設定的路徑。
2.2 Session和Session Time
不同於WEB應用中的Session，在資料分析中的Session會話，是指在指定的時間段內在網站上發生的一系列互動。本模型中的Session Time的含義是，當兩個行為間隔時間超過Session Time，我們便認為這兩個行為不屬於同一條路徑。
2.3 桑基圖
桑基圖（Sankey diagram），即桑基能量分流圖，也叫桑基能量平衡圖。它是一種特定型別的流程圖，圖中延伸的分支的寬度對應資料流量的大小。如圖4.1-1所示，每條邊表示上一節點到該節點的流量。一個完整的桑基圖包括以下幾個內容：節點資料及節點轉化率（下圖紅框部分）、邊資料及邊轉化率（下圖黑框部分）。轉化率的計算詳見【3.5. 轉化率計算】。
2.4 鄰接表
構造桑基圖可以簡化為一個圖的壓縮儲存問題。圖通常由幾個部分組成：
- 邊（edge）
- 點（vertex）
- 權重（weight）
- 度（degree）
本模型中，我們採用鄰接表進行儲存。鄰接表是一種常用的圖壓縮儲存結構，藉助連結串列來儲存圖中的節點和邊而忽略各節點之間不存在的邊，從而對矩陣進行壓縮。鄰接表的構造如下：
（a）中，左側為頂點節點，包含頂點資料及指向第一條邊的指標；右側為邊節點，包含該邊的權重、出入度等邊資訊以及指向下一條邊的指標。一個完整的鄰接表類似於Hashmap的結構，如圖(b)，左側是一個順序表，儲存的是(a)中的邊節點；每個邊節點對應一個連結串列儲存與該節點相連線的邊。頁面路徑模型中，為了適應模型的需要，我們對頂點節點和邊節點結構做了改造，詳情請見【4.1】節。
2.5 樹的剪枝
剪枝是樹的構造中一個重要的步驟，指刪去一些不重要的節點來降低計算或搜尋的複雜度。頁面路徑模型中，我們在剪枝環節對原始資料構造的樹進行修整，去掉不符合條件的分支，來保證樹中每條根節點到葉節點路徑的完整性。
2.6 PV和SV
PV即Page View，訪問次數，本模型中指的是一段時間內訪問的次數；SV即Session View，會話次數，本模型中指出現過該訪問路徑的會話數。如，有路徑一：A → B → C → D → A → B和路徑二：A → B → D，那麼，A → B的PV為2+1=3，SV為1+1=2。
三、 資料模型設計
本節將介紹資料模型的設計，包括資料流向、路徑劃分、ps/sv計算以及最終得到的桑基圖中路徑的轉化率計算。
3.1 整體資料流向
資料來源於統一的資料倉庫，通過Spark計算後寫入Clickhouse，並用Hive進行冷備份。資料流向圖見圖3.1-1。
圖3.1-1
3.2 技術選型
Clickhouse不是本文的重點，在此不詳細描述，僅簡要說明選擇Clickhouse的原因。
選擇的原因是在於，Clickhouse是列式儲存，速度極快。看下資料量級和查詢速度（截止到本文撰寫的日期）：
圖3.2-1
最後得到的千億資料查詢速度是這樣，
圖3.2-2
3.3 資料建模
3.3.1 獲取頁面資訊，劃分session
頁面路徑模型基於各種事件id切割獲取到對應的頁面id，來進行頁面路徑分析。Session的概念可見第2.2節，這裡不再贅述。目前我們使用更加靈活的Session劃分，使得使用者可以查詢到在各種時間粒度（5,10,15,30,60分鐘）的Session會話下，使用者的頁面轉化資訊。
假設有使用者a和使用者b，a使用者當天發生的行為事件分別為 E1, E2, E3... , 對應的頁面分別為P1, P2, P3... ，事件發生的時間分別為T1, T2, T3... ，選定的session間隔為tg。如圖所示T4-T3>tg，所以P1,P2,P3被劃分到了第一個Session，P4,P5被劃分到了第二個Session，同理P6及後面的頁面也被劃分到了新的Session。
虛擬碼實現如下：
def splitPageSessions(timeSeq: Seq[Long], events: Seq[String], interval: Int)
(implicit separator: String): Array[Array[Array[String]]] = {
// 引數中的events是事件集合，timeSeq是相應的事件發生時間的集合
if (events.contains(separator))
throw new IllegalArgumentException("Separator should't be in events.")
if (events.length != timeSeq.length)
throw new Exception("Events and timeSeq not in equal length.")
val timeBuf = ArrayBuffer[String](timeSeq.head.toString) // 儲存含有session分隔標識的時間集合
val eventBuf = ArrayBuffer[String](events.head) // 儲存含有session分隔標識的事件集合
if (timeSeq.length >= 2) {
events.indices.tail.foreach { i =>
if (timeSeq(i) - timeSeq(i - 1) > interval * 60000) { // 如果兩個事件的發生時間間隔超過設定的時間間隔，則新增分隔符作為後面劃分session的標識
timeBuf += separator;
eventBuf += separator
}
timeBuf += timeSeq(i).toString;
eventBuf += events(i)
}
}
val tb = timeBuf.mkString(",").split(s",\\$separator,").map(_.split(",")) // 把集合通過識別符號劃分成為各個session下的時間集合
val eb = eventBuf.mkString(",").split(s",\\$separator,").map(_.split(",")) // 把集合通過識別符號劃分成為各個session下的事件集合
tb.zip(eb).map(t => Array(t._1, t._2)) // 把session中的事件和發生時間對應zip到一起，並把元組修改成陣列型別，方便後續處理
}
3.3.2 相鄰頁面去重
不同的事件可能對應同一頁面，臨近的相同頁面需要被過濾掉，所以劃分session之後需要做的就是相鄰頁面去重。
圖3.3-2
相鄰頁面去重後得到的結果是這樣
圖3.3-3
3.3.3 獲取每個頁面的前/後四級頁面
然後對上述資料進行視窗函式分析，獲取每個session中每個頁面的前後四級頁面，其中sid是根據使用者標識ID和session號拼接而成，比如，針對上述的使用者a的第一個session 0會生成如下的7條記錄，圖中的page列為當前頁面，空頁面用-1表示
圖3.3-4
計算剩下的，會得到一共7+7+6+4+5=29條記錄。得到全部記錄如下
3.3.4 統計正負向路徑的pv/sv
取page和page_id_previous1, page_id_previous2, page_id_previous3 ,page_id_previous4得到負向五級路徑（path_direction為2），取page和page_id_next1, page_id_next2, page_id_next3, page_id_next4得到正向五級路徑（path_direction為1），分別計算路徑的pv和sv（按照sid去重），得到如下資料dfSessions，
直接看上面的資料可能比較茫然，所以這裡拆出兩條資料示例，第一條結果資料
這是一條正向的（path_direction為1）路徑結果資料，在下圖中就是從左到右的路徑，對應的兩個路徑如下
圖3.3-5
第二條結果資料
圖3.3-6
也是一條正向的路徑結果資料，其中pv為2，對應的兩個路徑如下，sv為1的原因是這兩條路徑的sid一致，都是使用者a在S1會話中產生的路徑
圖3.3-7
3.3.5 統計計算各級路徑的pv/sv
然後根據dfSessions資料，按照page_id_lv1分組計算pv和sv的和，得到一級路徑的pv和sv，一級路徑特殊地會把path_direction設定為0
然後類似地分別計算二三四五級路徑的pv和sv，合併所有結果得到如下
3.4 資料寫入
通過Spark分析計算的結果資料需要寫入Clickhouse來線上服務，寫入Hive來作為資料冷備份，可以進行Clickhouse的資料恢復。
Clickhouse表使用的是分散式（Distributed）表結構，分散式表本身不儲存任何資料，而是作為資料分片的透明代理，自動路由到資料到叢集中的各個節點，所以分散式表引擎需要配合其他資料表引擎一起使用。使用者路徑分析模型的表資料被儲存在叢集的各個分片中，分片方式使用隨機分片，在這裡涉及到了Clickhouse的資料寫入，我們展開講解下。
有關於這一點，在模型初期我們使用的是寫分散式表的方式來寫入資料，具體的寫入流程如下所示：
- 客戶端和叢集中的A節點建立jdbc連線，並通過HTTP的POST請求寫入資料；
- A分片在收到資料之後會做兩件事情，第一，根據分片規則劃分資料，第二，將屬於當前分片的資料寫入自己的本地表；
- A分片將屬於遠端分片的資料以分割槽為單位，寫入目錄下臨時bin檔案，命名規則如：/database@host:port/[increase_num].bin；
- A分片嘗試和遠端分片建立連線；
- 會有另一組監聽任務監聽上面產生的臨時bin檔案，並將這些資料傳送到遠端分片，每份資料單執行緒傳送；
- 遠端分片接收資料並且寫入本地表；
- A分片確認完成寫入。
通過以上過程可以看出，Distributed表負責所有分片的資料寫入工作，所以建立jdbc連線的節點的出入流量會峰值極高，會產生以下幾個問題：
- 單臺節點的負載過高，主要體現在記憶體、網絡卡出入流量和TCP連線等待數量等，機器健康程度很差；
- 當業務增長後更多的模型會接入Clickhouse做OLAP，意味著更大的資料量，以當前的方式來繼續寫入的必然會造成單臺機器宕機，在當前沒有做高可用的狀況下，單臺機器的宕機會造成整個叢集的不可用；
- 後續一定會做ck叢集的高可用，使用可靠性更高的ReplicatedMergeTree，使用這種引擎在寫入資料的時候，也會因為寫分散式表而出現數據不一致的情況。
針對於此資料端做了DNS輪詢寫本地表的改造，經過改造之後：
- 用於JDBC連線的機器的TCP連線等待數由90下降到25，降低了72%以上；
- 用於JDBC連線的機器的入流量峰值由645M/s降低到76M/s，降低了88%以上；
- 用於JDBC連線的機器因分發資料而造成的出流量約為92M/s，改造後這部分出流量清零。
另外，在Distributed表負責向遠端分片寫入資料的時候，有非同步寫和同步寫兩種方式，非同步寫的話會在Distributed表寫完本地分片之後就會返回寫入成功資訊，如果是同步寫，會在所有分片都寫入完成才返回成功資訊，預設的情況是非同步寫，我們可以通過修改引數來控制同步寫的等待超時時間。
def splitPageSessions(timeSeq: Seq[Long], events: Seq[String], interval: Int)
(implicit separator: String): Array[Array[Array[String]]] = {
// 引數中的events是事件集合，timeSeq是相應的事件發生時間的集合
if (events.contains(separator))
throw new IllegalArgumentException("Separator should't be in events.")
if (events.length != timeSeq.length)
throw new Exception("Events and timeSeq not in equal length.")
val timeBuf = ArrayBuffer[String](timeSeq.head.toString) // 儲存含有session分隔標識的時間集合
val eventBuf = ArrayBuffer[String](events.head) // 儲存含有session分隔標識的事件集合
if (timeSeq.length >= 2) {
events.indices.tail.foreach { i =>
if (timeSeq(i) - timeSeq(i - 1) > interval * 60000) { // 如果兩個事件的發生時間間隔超過設定的時間間隔，則新增分隔符作為後面劃分session的標識
timeBuf += separator;
eventBuf += separator
}
timeBuf += timeSeq(i).toString;
eventBuf += events(i)
}
}
val tb = timeBuf.mkString(",").split(s",\\$separator,").map(_.split(",")) // 把集合通過識別符號劃分成為各個session下的時間集合
val eb = eventBuf.mkString(",").split(s",\\$separator,").map(_.split(",")) // 把集合通過識別符號劃分成為各個session下的事件集合
tb.zip(eb).map(t => Array(t._1, t._2)) // 把session中的事件和發生時間對應zip到一起，並把元組修改成陣列型別，方便後續處理
}
3.5 轉化率計算
在前端頁面選擇相應的維度，選中起始頁面：
後端會在Clickhouse中查詢，
- 選定節點深度（node_depth）為1和一級頁面（page_id_lv1）是選定頁面的資料，得到一級頁面及其sv/pv，
- 選定節點深度（node_depth）為2和一級頁面（page_id_lv1）是選定頁面的資料，按照sv/pv倒序取前10，得到二級頁面及其sv/pv，
- 選定節點深度（node_depth）為2和一級頁面（page_id_lv1）是選定頁面的資料，按照sv/pv倒序取前20，得到三級頁面及其sv/pv，
- 選定節點深度（node_depth）為2和一級頁面（page_id_lv1）是選定頁面的資料，按照sv/pv倒序取前30，得到四級頁面及其sv/pv，
- 選定節點深度（node_depth）為2和一級頁面（page_id_lv1）是選定頁面的資料，按照sv/pv倒序取前50，得到五級頁面及其sv/pv，
轉化率計算規則：
頁面轉化率：
假設有路徑 A-B-C，A-D-C，A-B-D-C，其中ABCD分別是四個不同頁面
計算三級頁面C的轉化率：
(所有節點深度為3的路徑中三級頁面是C的路徑的pv/sv和)÷(一級頁面的pv/sv)
路徑轉化率
假設有A-B-C，A-D-C，A-B-D-C，其中ABCD分別是四個不同頁面
計算A-B-C路徑中B-C的轉化率：
(A-B-C這條路徑的pv/sv)÷(所有節點深度為3的路徑中二級頁面是B的路徑的pv/sv和)
四、工程端架構設計
本節將講解工程端的處理架構，包括幾個方面：桑基圖的構造、路徑合併以及轉化率計算、剪枝。
4.1 桑基圖的構造
從上述原型圖可以看到，我們需要構造桑基圖，對於工程端而言就是需要構造帶權路徑樹。
簡化一下上圖，就可以將需求轉化為構造帶權樹的鄰接表。如下左圖就是我們的鄰接表設計。左側順序列表儲存的是各個節點（Vertex），包含節點名稱（name）、節點程式碼（code）等節點資訊和一個指向邊（Edge）列表的指標；每個節點（Vertex）指向一個邊（Edge）連結串列，每條邊儲存的是當前邊的權重、端點資訊以及指向同節點下一條邊的指標。
圖4.1-2
圖4.1-3
圖4.1-2就是我們在模型中使用到的鄰接表。這裡在2.4中描述的鄰接表上做了一些改動。在我們的桑基圖中，不同層級會出現相同名稱不同轉化率的節點，這些節點作為路徑的一環，並不能按照名稱被看作重複節點，不構成環路。如果整個桑基圖用一個鄰接表表示，那麼這類節點將被當作相同節點，使得影象當中出現環路。因此，我們將桑基圖按照層級劃分，每兩級用一個鄰接表表示，如圖4.1-2，Level 1表示層級1的節點和指向層級2的邊、Level 2表示層級2的節點指向層級3的邊，以此類推。
4.2 路徑的定義
首先，我們先回顧一下桑基圖：
觀察上圖可以發現，我們需要計算四個資料：每個節點的pv/sv、每個節點的轉化率、節點間的pv/sv、節點間的轉化率。那麼下面我們給出這幾個資料的定義：
- 節點pv/sv = 當前節點在當前層次中的pv/sv總和
- 節點轉化率 = ( 節點pv/sv ) / ( 路徑起始節點pv/sv )
- 節點間pv/sv = 上一級節點流向當前節點的pv/sv
- 節點間轉化率 = ( 節點間pv/sv ) / ( 上一級節點pv/sv )
再來看下儲存在Clickhouse中的路徑資料。先來看看錶結構：
( `node_depth` Int8 COMMENT '節點深度，共5個層級深度，列舉值1-2-3-4-5' CODEC(T64, LZ4HC(0)), `page_id_lv1` String COMMENT '一級頁面，起始頁面' CODEC(LZ4HC(0)), `page_id_lv2` String COMMENT '二級頁面' CODEC(LZ4HC(0)), `page_id_lv3` String COMMENT '三級頁面' CODEC(LZ4HC(0)), `page_id_lv4` String COMMENT '四級頁面' CODEC(LZ4HC(0)), `page_id_lv5` String COMMENT '五級頁面' CODEC(LZ4HC(0)) )
上述為路徑表中比較重要的幾個欄位，分別表示節點深度和各級節點。表中的資料包含了完整路徑和中間路徑。完整路徑指的是：路徑從起點到退出、從起點到達指定終點，超出5層的路徑當作5層路徑來處理。中間路徑是指資料計算過程中產生的中間資料，並不能作為一條完整的路徑。
路徑資料：
（1）完整路徑
（2）不完整路徑
那麼我們需要從資料中篩選出完整路徑，並將路徑資料組織成樹狀結構。
4.3 設計實現
4.3.1 整體框架
後端整體實現思路很明確，主要步驟就是讀取資料、構造鄰接表和剪枝。那麼要怎麼實現完整/非完整路徑的篩選呢？我們通過service層剪枝來過濾掉不完整的路徑。以下是描述整個流程的虛擬碼：
// 1-1: 分層讀取原始資料
// 1-1-1: 分層構造Clickhouse Sql
for( int depth = 1; depth <= MAX_DEPTH; depth ++){
sql.append(select records where node_depth = depth)
}
// 1-1-2: 讀取資料
clickPool.getClient();
records = clickPool.getResponse(sql);
// 2-1: 獲取節點之間的父子、子父關係（雙向edge構造）
findFatherAndSonRelation(records);
findSonAndFathRelation(records);
// 3-1: 剪枝
// 3-1-1: 清除孤立節點
for(int depth = 2; depth <= MAX_DEPTH; depth ++){
while(hasNode()){
node = getNode();
if node does not have father in level depth-1:
cut out node;
}
}
// 3-1-2: 過濾不完整路徑
for(int depth = MAX_DEPTH - 1; depth >= 1; depth --){
cut out this path;
}
// 3-2: 構造鄰接表
while(node.hasNext()){
sumVal = calculate the sum of pv/sv of this node until this level;
edgeDetails = get the details of edges connected to this node and the end point connected to the edges;
sortEdgesByEndPoint(edgeDetails);
path = new Path(sumVal, edgeDetails);
}
4.3.2 Clickhouse連線池
頁面路徑中我們引入了ClickHouse，其特點在這裡不再贅述。我們使用一個簡單的Http連線池連線ClickHouse Server。連線池結構如下：
4.3.3 資料讀取
如2中描述的，我們需要讀取資料中的完整路徑。
( `node_depth` Int8 COMMENT '節點深度，列舉值', `page_id_lv1` String COMMENT '一級頁面，起始頁面', `page_id_lv2` String COMMENT '二級頁面', `page_id_lv3` String COMMENT '三級頁面', `page_id_lv4` String COMMENT '四級頁面', `page_id_lv5` String COMMENT '五級頁面', `val` Int64 COMMENT '全量資料value' )
在上述表結構中可以看到，寫入資料庫的路徑已經是經過一級篩選，深度≤5的路徑。我們需要在此基礎上再將完整路徑和不完整路徑區分開，根據需要根據node_depth和page_id_lvn來判斷是否為完整路徑並計算每個節點的value。
完整路徑判斷條件：
- node_depth=n, page_id_lvn=pageId (n < MAX_DEPTH)
- node_depth=n, page_id_lvn=pageId || page_id_lvn=EXIT_NODE (n = MAX_DEPTH)
完整路徑的條件我們已經知道了，那麼讀取路徑時有兩種方案。方案一：直接根據上述條件進行篩選來獲取完整路徑，由於Clickhouse及後端效能的限制，取數時必須limit；方案二：逐層讀取，可以計算全量資料，但是無法保證取出準確數量的路徑。
通過觀察發現，資料中會存在重複路徑，並且假設有兩條路徑：
A → B → C → D → EXIT_NODE
A → B → E → D → EXIT_NODE
當有以上兩條路徑時，需要計算每個節點的value。而在實際資料中，我們只能通過不完整路徑來獲取當前節點的value。因此，方案一不適用。
那麼方案二就可以通過以下虛擬碼逐層讀取：
for(depth = 1; depth <= MAX_DEPTH; depth++){
select
node_depth as nodeDepth,
...,
sum(sv) as val
from
table_name
where
...
AND (toInt16OrNull(pageId1) = 45)
AND (node_depth = depth)
...
group by
node_depth,
pageId1,
pageId2,
...
ORDER BY
...
LIMIT
...
}
讀取出的資料如下：
那麼，node1_A_val = 10+20，node2_B_val = 9+15 以此類推。
4.3.4 剪枝
根據4.3.3，在取數階段我們會分層取出所有原始資料，而原始資料中包含了完整和非完整路徑。如下圖是直接根據原始資料構造的樹（原始樹）。按照我們對完整路徑的定義：路徑深度達到5且結束節點為退出或其它節點；路徑深度未達到5且結束節點為退出。可見，圖中標紅的部分（node4_lv1 → node3_lv2）是一條不完整路徑。
另外，原始樹中還會出現孤立節點（綠色節點node4_lv2）。這是由於在取數階段，我們會對資料進行分層排序再取出，這樣一來無法保證每層資料的關聯性。因此，node4_lv2節點在lv2層排序靠前，而其前驅、後繼節點排序靠後無法選中，從而導致孤立節點產生。
圖4.3-3
因此，在我們取出原始資料集後，還需要進行過濾才能獲取我們真正需要的路徑。
在模型中，我們通過剪枝來實現這一過濾操作。
// 清除孤立節點
for(int depth = 2; depth <= MAX_DEPTH; depth ++){
while(hasNode()){
node = getNode();
if node does not have any father and son: // [1]
cut out node;
}
}
// 過濾不完整路徑
for(int depth = MAX_DEPTH - 1; depth >= 1; depth --){
cut out this path; // [2]
}
在前述的步驟中，我們已經獲取了雙向edge列表（父子關係和子父關係列表）。因此在上述虛擬碼[1]中，藉助edge列表即可快速查詢當前節點的前驅和後繼，從而判斷當前節點是否為孤立節點。
同樣，我們利用edge列表對不完整路徑進行裁剪。對於不完整路徑，剪枝時只需要關心深度不足MAX_DEPTH且最後節點不為EXIT_NODE的路徑。那麼在上述虛擬碼[2]中，我們只需要判斷當前層的節點是否存在順序邊（父子關係）即可，若不存在，則清除當前節點。
五、寫在最後
基於平臺化查詢中查詢時間短、需要視覺化的要求，並結合現有的儲存計算資源以及具體需求，我們在實現中將路徑資料進行列舉後分為兩次進行合併，第一次是同一天內對相同路徑進行合併，第二次是在日期區間內對路徑進行彙總。本文希望能為路徑分析提供參考，在使用時還需結合各業務自身的特性進行合理設計，更好地服務於業務。
方案中涉及到的Clickhouse在這裡不詳細介紹，感興趣的同學可以去深入瞭解，歡迎和筆者一起探討學習。
作者：vivo 網際網路大資料團隊