以flink實時流的方式實現OneId

前言

oneid相關概念,及其當前離線實現方式介紹請見以下鏈接及其系列其他文章:

用戶標簽(一):圖計算實現ID_Mapping、Oneid打通數據孤島

讀完上述鏈接的相關文章,將理解oneid的需求及其實現方式.

背景

之前公司是做電商數據分析,可以接入多個數據源的數據(美團,餓了麽,京東等),在我們系統中,我們將為每個用戶統一打上在我們系統內部的user_id,即oneid.當時已經有了基於spark圖計算實現的id-mapping來達成oneid.

到這裡,我們已有的系統跟前言中鏈接文章提到的內容相似.

後來,我們有了新需求:實時化改造.實時為每個用戶統計相關數據.

但是,實時計算的數據最後要歸納到用戶上,那麼我們的id-mapping也需要實時出現結果.

思考

本人不具備機器學習領域的知識,在看了幾遍前言中的文章後,勉強搞懂了該圖計算的原理,本質上是求最大連通圖,研究後發現需要拿到全量數據進行迭代,不能改造為實時計算.

後來,基於"最大連通圖"的演算法,變種出了一種方法:

1. 輸入兩個數據關聯關係,例如 美團id1-餓了麽id1,即輸入兩個點一個線.
2. 拿到該關係,分別拿兩個數據去與已有的id-mapping結果表對比.如,我們結果id-mapping表中,有關係

   美團id1 - oneid1
   餓了麽id1 - oneid2
   

3. 通過第一步中傳入的關係,可以得出 美團id1 與 餓了麽id1 在我們系統中應該識別為同一個人,對應同一個oneid,可以得到

   美團id1 - oneid1
   餓了麽id1 - oneid1
   

   或者,都對應為另一個oneid

   美團id1 - oneid2
   餓了麽id1 - oneid2
   

   即,我們通過傳入關聯關係,將 美團id1 與 餓了麽id1 在我們系統中重新更新為關聯到同一個oneid.
4. 後續,某個用戶id是 美團id1,那麼它關聯這個id-mapping結果表,可以得到它在我們系統中的id是oneid1(或者oneid2,此處根據第三步如何取值)
5. 根據不同的對比結果,進行相應的替換或者新增,我們變相實現了"最大連通圖"的演算法,並且這個演算法可以用flink實時計算實現

詳細步驟

0. id-mapping結果表設計

結果表可以有多個描述欄位,但是核心應該是以下兩個欄位:

原id , 計算出的oneid

1. 輸入數據採集

我們在採集數據的時候,需要將數據解析成兩兩的關係對.如原始數據:

手機號1,美團id1,設備id1

需要將這條消息拆分為:

手機號1 - 美團id1
手機號1 - 設備id1
美團id1 - 設備id1

再將這三組關聯關係傳給後續對比計算.

2. 對比計算

假設我們得到關係對:

x - y

我們拿到此關係對到結果表中進行對比將有以下幾種情況:

1. x,y都沒有對應oneid: 直接對結果表插入計算得出的新oneid(可以使用uuid)

   x - 新oneid
   y - 新oneid
   

2. x已有對應oneid為 XXoneid,y沒有:將y的oneid賦值為 XXoneid,並插入,得到

   x - XXoneid
   y - XXoneid
   

3. x沒有,y有oneid為 YYoneid :同第二種情況,得到

   x - YYoneid
   y - YYoneid
   

4. x,y都有oneid,且一致,都為 ZZoneid: 不更新
5. x,y都有oneid,且不一致,分別為 XXoneid,YYoneid :將 x,y更新為同一個oneid(XXoneid或者YYoneid),或者重新生成一個.此處看個人選擇.

   並且!!!!!!
   將結果表中所有oneid為 XXoneid,YYoneid的相關數據,oneid都重設為新選擇的oneid
   這是為了將相關聯的其他數據一起指向新的oneid
   

至此,通過以上幾種情況.我們復現了id-mapping中求最大連通圖的演算法.

實現程式設計

1. 數據源

kafka

2. 實時計算程式

flink

3. 對比中如何取數

redis:將結果表以k-v的形式放在記憶體中,這樣flink可以快速取值並對比計算

4. 結果表存放

hbase:此處可以換為mysql,doris等支持更新的存儲即可.並且還有以下原因:

對比計算中,第五種情況,需要從這裡取所有oneid為 XXoneid,YYoneid的相關數據
而redis中沒法根據value來取得key,所以第五種情況,需要查詢此處存儲得到相關數據

5. 結果更新

結果不但要更新hbase,還要更新redis中存放的k-v對!!!建議先更新redis,因為比較快.