以flink實時流的方式實作OneId

前言

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

用戶標簽(一):圖計算實作ID_Mapping、Oneid打通資料孤島

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

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

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

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

但是,實時計算的資料最后要歸納到用戶上,那么我們的id-mapping也需要實時出現結果.

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

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

輸入兩個資料關聯關系,例如美團id1-餓了么id1,即輸入兩個點一個線.
拿到該關系,分別拿兩個資料去與已有的id-mapping結果表對比.如,我們結果id-mapping表中,有關系
```
美團id1 - oneid1
餓了么id1 - oneid2
```
通過第一步中傳入的關系,可以得出美團id1 與餓了么id1 在我們系統中應該識別為同一個人,對應同一個oneid,可以得到
```
美團id1 - oneid1
餓了么id1 - oneid1
```
或者,都對應為另一個oneid
```
美團id1 - oneid2
餓了么id1 - oneid2
```
即,我們通過傳入關聯關系,將美團id1 與餓了么id1 在我們系統中重新更新為關聯到同一個oneid.
后續,某個用戶id是美團id1,那么它關聯這個id-mapping結果表,可以得到它在我們系統中的id是oneid1(或者oneid2,此處根據第三步如何取值)
根據不同的對比結果,進行相應的替換或者新增,我們變相實作了"最大連通圖"的演算法,并且這個演算法可以用flink實時計算實作

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

原id , 計算出的oneid

我們在采集資料的時候,需要將資料決議成兩兩的關系對.如原始資料:

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

需要將這條訊息拆分為:

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

再將這三組關聯關系傳給后續對比計算.

假設我們得到關系對:

x - y

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

x,y都沒有對應oneid: 直接對結果表插入計算得出的新oneid(可以使用uuid)
```
x - 新oneid
y - 新oneid
```
x已有對應oneid為 XXoneid,y沒有:將y的oneid賦值為 XXoneid,并插入,得到
```
x - XXoneid
y - XXoneid
```
x沒有,y有oneid為 YYoneid :同第二種情況,得到
```
x - YYoneid
y - YYoneid
```
x,y都有oneid,且一致,都為 ZZoneid: 不更新

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

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

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

kafka

flink

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

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

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

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

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/542567.html

標籤：其他