數(shù)倉(cāng)建?！狾neID

今天是我在上海租房的小區(qū)被封的第三天，由于我的大意，沒有屯吃的，外賣今天完全點(diǎn)不到了，中午的時(shí)候我找到了一包快過(guò)期的肉松餅，補(bǔ)充了1000焦耳的能量。但是中午去做核酸的時(shí)候，我感覺走路有點(diǎn)不穩(wěn)，我看到大白的棉簽深入我的嘴里，我竟然以為是吃的，差點(diǎn)咬住了，還好我有僅存的一點(diǎn)意識(shí)。下午我收到女朋友給我點(diǎn)的外賣——面包（我不知道她是怎么點(diǎn)到的外賣，我很感動(dòng)），很精致的面包，擱平時(shí)我基本不喜歡吃面包，但是已經(jīng)到了這個(gè)份上，我大口吃起來(lái)，竟然覺得這是世界上最好吃的食物了。明天早晨5：50的鬧鐘，去叮咚和美團(tuán)買菜，看能不能搶幾桶泡面吧。愿神保佑，我暗暗下著決心并祈禱著，胸前畫著十字。。。

數(shù)據(jù)倉(cāng)庫(kù)系列文章(持續(xù)更新)

1. 數(shù)倉(cāng)架構(gòu)發(fā)展史
2. 數(shù)倉(cāng)建模方法論
3. 數(shù)倉(cāng)建模分層理論
4. 數(shù)倉(cāng)建模—寬表的設(shè)計(jì)
5. 數(shù)倉(cāng)建?！笜?biāo)體系
6. 數(shù)據(jù)倉(cāng)庫(kù)之拉鏈表
7. 數(shù)倉(cāng)—數(shù)據(jù)集成
8. 數(shù)倉(cāng)—數(shù)據(jù)集市
9. 數(shù)倉(cāng)—商業(yè)智能系統(tǒng)
10. 數(shù)倉(cāng)—埋點(diǎn)設(shè)計(jì)與管理
11. 數(shù)倉(cāng)—ID Mapping
12. 數(shù)倉(cāng)—OneID
13. 數(shù)倉(cāng)—AARRR海盜模型
14. 數(shù)倉(cāng)—總線矩陣
15. 數(shù)倉(cāng)—數(shù)據(jù)安全
16. 數(shù)倉(cāng)—數(shù)據(jù)質(zhì)量
17. 數(shù)倉(cāng)—數(shù)倉(cāng)建模和業(yè)務(wù)建模

OneID

前面我們學(xué)習(xí)了ID Mapping，包括ID Mapping 的背景介紹和業(yè)務(wù)場(chǎng)景，以及如何使用Spark 實(shí)現(xiàn)ID Mapping，這個(gè)過(guò)程中涉及到了很多東西，當(dāng)然我們都通過(guò)文章的形式介紹給大家了，所以你再學(xué)習(xí)今天這一節(jié)之前，可以先看一下前面的文章

1. Spark實(shí)戰(zhàn)—GraphX編程指南
2. 數(shù)倉(cāng)建模—ID Mapping

在上一節(jié)我們介紹ID Mapping 的時(shí)候我們就說(shuō)過(guò)ID Mapping ?是為了打通用戶各個(gè)維度的數(shù)據(jù)，從而消除數(shù)據(jù)孤島、避免數(shù)據(jù)歧義，從而更好的刻畫用戶，所以說(shuō)ID Mapping是手段不是目的，目的是為了打通數(shù)據(jù)體系，ID Mapping最終的產(chǎn)出就是我們今天的主角OneID，也就是說(shuō)數(shù)據(jù)收集過(guò)來(lái)之后通過(guò)ID Mapping 打通，從而產(chǎn)生OneID，這一步之后我們的整個(gè)數(shù)據(jù)體系就將使用OneID作為用戶的ID，這樣我們整個(gè)數(shù)據(jù)體系就得以打通

OneData

開始之前我們先看一下阿里的OneData 數(shù)據(jù)體系，從而更好認(rèn)識(shí)一下OneID，前面我們說(shuō)過(guò)ID Mapping 只是手段不是目的，目的是為了打通數(shù)據(jù)體系，ID Mapping最終的產(chǎn)出就是OneID

其實(shí)OneID在我們整個(gè)數(shù)據(jù)服務(wù)體系中，也只是起點(diǎn)不是終點(diǎn)或者說(shuō)是手段，我們最終的目的是為了建設(shè)統(tǒng)一的數(shù)據(jù)資產(chǎn)體系。

沒有建設(shè)統(tǒng)一的數(shù)據(jù)資產(chǎn)體系之前，我們的數(shù)據(jù)體系建設(shè)存在下面諸多問(wèn)題

1. 數(shù)據(jù)孤島：各產(chǎn)品、業(yè)務(wù)的數(shù)據(jù)相互隔離，難以通過(guò)共性ID打通
2. 重復(fù)建設(shè)：重復(fù)的開發(fā)、計(jì)算、存儲(chǔ)，帶來(lái)高昂的數(shù)據(jù)成本
3. 數(shù)據(jù)歧義：指標(biāo)定義口徑不一致，造成計(jì)算偏差，應(yīng)用困難

在阿里巴巴 OneData 體系中，OneID 指統(tǒng)一數(shù)據(jù)萃取，是一套解決數(shù)據(jù)孤島問(wèn)題的思想和方法。數(shù)據(jù)孤島是企業(yè)發(fā)展到一定階段后普遍遇到的問(wèn)題。各個(gè)部門、業(yè)務(wù)、產(chǎn)品，各自定義和存儲(chǔ)其數(shù)據(jù)，使得這些數(shù)據(jù)間難以關(guān)聯(lián)，變成孤島一般的存在。

OneID的做法是通過(guò)統(tǒng)一的實(shí)體識(shí)別和連接，打破數(shù)據(jù)孤島，實(shí)現(xiàn)數(shù)據(jù)通融。簡(jiǎn)單來(lái)說(shuō)，用戶、設(shè)備等業(yè)務(wù)實(shí)體，在對(duì)應(yīng)的業(yè)務(wù)數(shù)據(jù)中，會(huì)被映射為唯一識(shí)別（UID）上，其各個(gè)維度的數(shù)據(jù)通過(guò)這個(gè)UID進(jìn)行關(guān)聯(lián)。

各個(gè)部門、業(yè)務(wù)、產(chǎn)品對(duì)業(yè)務(wù)實(shí)體的UID的定義和實(shí)現(xiàn)不一樣，使得數(shù)據(jù)間無(wú)法直接關(guān)聯(lián)，成為了數(shù)據(jù)孤島?；谑謾C(jī)號(hào)、身份證、郵箱、設(shè)備ID等信息，結(jié)合業(yè)務(wù)規(guī)則、機(jī)器學(xué)習(xí)、圖算法等算法，進(jìn)行 ID-Mapping，將各種 UID 都映射到統(tǒng)一ID上。通過(guò)這個(gè)統(tǒng)一ID，便可關(guān)聯(lián)起各個(gè)數(shù)據(jù)孤島的數(shù)據(jù)，實(shí)現(xiàn)數(shù)據(jù)通融，以確保業(yè)務(wù)分析、用戶畫像等數(shù)據(jù)應(yīng)用的準(zhǔn)確和全面。

OneModel 統(tǒng)一數(shù)據(jù)構(gòu)建和管理

將指標(biāo)定位細(xì)化為：

1. 原子指標(biāo)
2. 時(shí)間周期
3. 修飾詞（統(tǒng)計(jì)粒度、業(yè)務(wù)限定, etc）

通過(guò)這些定義，設(shè)計(jì)出各類派生指標(biāo) 基于數(shù)據(jù)分層，設(shè)計(jì)出維度表、明細(xì)事實(shí)表、匯總事實(shí)表，其實(shí)我們看到OneModel 其實(shí)沒有什么新的內(nèi)容，其實(shí)就是我們數(shù)倉(cāng)建模的那一套東西

OneService 統(tǒng)一數(shù)據(jù)服務(wù)

OneService 基于復(fù)用而不是復(fù)制數(shù)據(jù)的思想，指得是我們的統(tǒng)一的數(shù)據(jù)服務(wù)，因?yàn)槲覀円恢痹偬岢珡?fù)用，包括我們數(shù)倉(cāng)的建設(shè)，但是我們的數(shù)據(jù)服務(wù)這一塊卻是空白，所以O(shè)neService核心是服務(wù)的復(fù)用，能力包括：

• 利用主題邏輯表屏蔽復(fù)雜物理表的主題式數(shù)據(jù)服務(wù)
• 一般查詢+ OLAP 分析+在線服務(wù)的統(tǒng)一且多樣化數(shù)據(jù)服務(wù)
• 屏蔽多種異構(gòu)數(shù)據(jù)源的跨源數(shù)據(jù)服務(wù)

OneID 統(tǒng)一數(shù)據(jù)萃取

基于統(tǒng)一的實(shí)體識(shí)別、連接和標(biāo)簽生產(chǎn)，實(shí)現(xiàn)數(shù)據(jù)通融，包括：

• ID自動(dòng)化識(shí)別與連接
• 行為元素和行為規(guī)則
• 標(biāo)簽生產(chǎn)

OneID基于超強(qiáng)ID識(shí)別技術(shù)鏈接數(shù)據(jù)，高效生產(chǎn)標(biāo)簽；業(yè)務(wù)驅(qū)動(dòng)技術(shù)價(jià)值化，消除數(shù)據(jù)孤島，提升數(shù)據(jù)質(zhì)量，提升數(shù)據(jù)價(jià)值。

而ID的打通，必須有ID-ID之間的兩兩映射打通關(guān)系，通過(guò)ID映射關(guān)系表，才能將多種ID之間的關(guān)聯(lián)打通，完全孤立的兩種ID是無(wú)法打通的。

打通整個(gè)ID體系，看似簡(jiǎn)單，實(shí)則計(jì)算復(fù)雜，計(jì)算量非常大。假如某種對(duì)象有數(shù)億個(gè)個(gè)體，每個(gè)個(gè)體又有數(shù)十種不同的ID標(biāo)識(shí)，任意兩種ID之間都有可能打通關(guān)系，想要完成這類對(duì)象的所有個(gè)體ID打通需要數(shù)億次計(jì)算，一般的機(jī)器甚至大數(shù)據(jù)集群都無(wú)法完成。

大數(shù)據(jù)領(lǐng)域中的ID-Mapping技術(shù)就是用機(jī)器學(xué)習(xí)算法類來(lái)取代野蠻計(jì)算，解決對(duì)象數(shù)據(jù)打通的問(wèn)題?；谳斎氲腎D關(guān)系對(duì)，利用機(jī)器學(xué)習(xí)算法做穩(wěn)定性和收斂性計(jì)算，輸出關(guān)系穩(wěn)定的ID關(guān)系對(duì)，并生成一個(gè)UID作為唯一識(shí)別該對(duì)象的標(biāo)識(shí)碼。

OneID實(shí)現(xiàn)過(guò)程中存在的問(wèn)題

前面我們知道我們的ID Mapping 是通過(guò)圖計(jì)算實(shí)現(xiàn)，核心就是連通圖，其實(shí)實(shí)現(xiàn)OneID我們?cè)诖蛲↖D 之后，我們就可以為一個(gè)個(gè)連通圖生成一個(gè)ID, 因?yàn)橐粋€(gè)連通圖 就代表一個(gè)用戶，這樣我們生成的ID就是用戶的OneID，這里的用戶指的是自然人，而不是某一個(gè)平臺(tái)上的用戶。

OneID 的生成問(wèn)題

首先我們需要一個(gè)ID 生成算法，因?yàn)槲覀冃枰獮榇罅坑脩羯蒊D,我們的ID 要求是唯一的，所以在算法設(shè)計(jì)的時(shí)候就需要考慮到這一點(diǎn)，我們并不推薦使用UUID,原因是UUID了可能會(huì)出現(xiàn)重復(fù)，而且UUID 沒有含義，所以我們不推薦使用UUID,我們這里使用的是MD5 算法，所以我們的MD5 算法的參數(shù)是我們的圖的標(biāo)示ID。

OneID 的更新問(wèn)題

這里的更新問(wèn)題主要就是我們的數(shù)據(jù)每天都在更新，也就是說(shuō)我們的圖關(guān)系在更新，也就是說(shuō)我們要不要給這個(gè)自然人重新生成OneID ,因?yàn)樗膱D關(guān)系可能發(fā)生了變化。

其實(shí)這里我們不能為該自然人生成新的OneID ，否則我們數(shù)倉(cāng)里的歷史數(shù)據(jù)可能無(wú)法關(guān)聯(lián)使用，所以我們的策略就是如果該自然人已經(jīng)有OneID了，則不需要重新生成，其實(shí)這里我們就是判斷該圖中的所有的頂點(diǎn)是否存在OneID，我們后面在代碼中體現(xiàn)著一點(diǎn)。

OneID 的選擇問(wèn)題

這個(gè)和上面的更新問(wèn)題有點(diǎn)像，上面更新問(wèn)題我們可以保證一個(gè)自然人的OneID不發(fā)生變化，但是選擇問(wèn)題會(huì)導(dǎo)致發(fā)生變化，但是這個(gè)問(wèn)題是圖計(jì)算中無(wú)法避免的，我們舉個(gè)例子，假設(shè)我們有用戶的兩個(gè)ID(A_ID,C_ID)，但是這兩個(gè)ID 在當(dāng)前是沒有辦法打通的，所以我們就會(huì)為這個(gè)兩個(gè)ID 生成兩個(gè)OneID,也就是(A_OneID,B_OneID)，所以這個(gè)時(shí)候我們知道因?yàn)镮D Mapping 不上，所以我們認(rèn)為這兩個(gè)ID 是兩個(gè)人。

后面我們有了另外一個(gè)ID(B_ID),這個(gè)ID可以分別和其他的兩個(gè)ID 打通，也就是B_ID<——>A_ID , B_ID<——>C_ID 這樣我們就打通這個(gè)三個(gè)ID，這個(gè)時(shí)候我們知道

這個(gè)用戶存在三個(gè)ID,并且這個(gè)時(shí)候已經(jīng)存在了兩個(gè)OneID，所以這個(gè)時(shí)候我們需要在這兩個(gè)OneID中選擇一個(gè)作為用戶的OneID，簡(jiǎn)單粗暴點(diǎn)就可以選擇最小的或者是最大的。

我們選擇了之后，要將另外一個(gè)OneID對(duì)應(yīng)的數(shù)據(jù)，對(duì)應(yīng)到選擇的OneID 下，否則沒有被選擇的OneID的歷史數(shù)據(jù)就無(wú)法追溯了

OneID 代碼實(shí)現(xiàn)

這個(gè)代碼相比ID Mapping主要是多了OneID 的生成邏輯和更新邏輯 ，需要注意的是關(guān)于頂點(diǎn)集合的構(gòu)造我們不是直接使用字符串的hashcode ，這是因?yàn)閔ashcode 很容易重復(fù)

object?OneID??{
????val?spark?=?SparkSession
??????.builder()
??????.appName("OneID")
??????.getOrCreate()

??val?sc?=?spark.sparkContext

??def?main(args:?Array[String]):?Unit?=?{
????val?bizdate=args(0)
????val?c?=?Calendar.getInstance
????val?format?=?new?SimpleDateFormat("yyyyMMdd")
????c.setTime(format.parse(bizdate))

????c.add(Calendar.DATE,?-1)
????val?bizlastdate?=?format.format(c.getTime)

????println(s"?時(shí)間參數(shù)??${bizdate}????${bizlastdate}")
????//?dwd_patient_identity_info_df?就是我們用戶的各個(gè)ID?，也就是我們的數(shù)據(jù)源
????//?獲取字段，這樣我們就可以擴(kuò)展新的ID?字段，但是不用更新代碼
????val?columns?=?spark.sql(
??????s"""
?????????|select
?????????|???*
?????????|from
?????????|???lezk_dw.dwd_patient_identity_info_df
?????????|where
?????????|???ds='${bizdate}'
?????????|limit
?????????|???1
?????????|""".stripMargin)
??????.schema.fields.map(f?=>?f.name).filterNot(e=>e.equals("ds")).toList

????//?獲取數(shù)據(jù)
????val?dataFrame?=?spark.sql(
??????s"""
????????|select
????????|???${columns.mkString(",")}
????????|from
????????|???lezk_dw.dwd_patient_identity_info_df
????????|where
????????|???ds='${bizdate}'
????????|""".stripMargin
????)

????//?數(shù)據(jù)準(zhǔn)備
????val?data?=?dataFrame.rdd.map(row?=>?{
??????val?list?=?new?ListBuffer[String]()
??????for?(column?<-?columns)?{
????????val?value?=?row.getAs[String](column)
????????list.append(value)
??????}
??????list.toList
????})
????import?spark.implicits._
????//?頂點(diǎn)集合
????val?veritx=?data.flatMap(list?=>?{
??????for?(i?<-?0?until?columns.length?if?StringUtil.isNotBlank(list(i))?&&?(!"null".equals(list(i))))
????????yield?(new?BigInteger(DigestUtils.md5Hex(list(i)),16).longValue,?list(i))

????}).distinct

????val?veritxDF=veritx.toDF("id_hashcode","id")
????veritxDF.createOrReplaceTempView("veritx")

????//?生成邊的集合
????val?edges?=?data.flatMap(list?=>?{
??????for?(i?<-?0?to?list.length?-?2?if?StringUtil.isNotBlank(list(i))?&&?(!"null".equals(list(i)))
???????????;?j?<-?i?+?1?to?list.length?-?1?if?StringUtil.isNotBlank(list(j))?&&?(!"null".equals(list(j))))
??????yield?Edge(new?BigInteger(DigestUtils.md5Hex(list(i)),16).longValue,new?BigInteger(DigestUtils.md5Hex(list(j)),16).longValue,?"")
????}).distinct


????//?開始使用點(diǎn)集合與邊集合進(jìn)行圖計(jì)算訓(xùn)練
????val?graph?=?Graph(veritx,?edges)
????val?connectedGraph=graph.connectedComponents()

????//?連通節(jié)點(diǎn)
????val??vertices?=?connectedGraph.vertices.toDF("id_hashcode","guid_hashcode")
????vertices.createOrReplaceTempView("to_graph")

????//?加載昨日的oneid?數(shù)據(jù)?(oneid,id,id_hashcode)?
????val?ye_oneid?=?spark.sql(
??????s"""
????????|select
????????|???oneid,id,id_hashcode
????????|from
????????|???lezk_dw.dwd_patient_oneid_info_df
????????|where
????????|???ds='${bizlastdate}'
????????|""".stripMargin
????)
????ye_oneid.createOrReplaceTempView("ye_oneid")

????//?關(guān)聯(lián)獲取?已經(jīng)存在的?oneid，這里的min?函數(shù)就是我們說(shuō)的oneid?的選擇問(wèn)題
????val?exists_oneid=spark.sql(
??????"""
????????|select
????????|???a.guid_hashcode,min(b.oneid)?as?oneid
????????|from
????????|???to_graph?a
????????|inner?join
????????|???ye_oneid?b
????????|on
????????|???a.id_hashcode=b.id_hashcode
????????|group?by
????????|???a.guid_hashcode
????????|""".stripMargin
????)
????exists_oneid.createOrReplaceTempView("exists_oneid")
????//?不存在則生成?存在則取已有的?這里nvl?就是oneid??的更新邏輯，存在則獲取?不存在則生成
????val?today_oneid=spark.sql(
??????s"""
????????|insert?overwrite?table?dwd_patient_oneid_info_df?partition(ds='${bizdate}')
????????|select
????????|???nvl(b.oneid,md5(cast(a.guid_hashcode?as?string)))?as?oneid,c.id,a.id_hashcode,d.id?as?guid,a.guid_hashcode
????????|from
????????|???to_graph?a
????????|left?join
????????|???exists_oneid?b
????????|on
????????|???a.guid_hashcode=b.guid_hashcode
????????|left?join
????????|???veritx?c
????????|on
????????|???a.id_hashcode=c.id_hashcode
????????|left?join
????????|???veritx?d
????????|on
????????|???a.guid_hashcode=d.id_hashcode
????????|""".stripMargin
????)
????sc.stop
??}

}

這個(gè)代碼中我們使用了SparkSQL,其實(shí)你如果更加擅長(zhǎng)RDD的API，也可以使用RDD 優(yōu)化，需要注意的是網(wǎng)上的很多代碼中使用了廣播變量，將vertices 變量廣播了出去，其實(shí)這個(gè)時(shí)候存在一個(gè)風(fēng)險(xiǎn)那就是如果你的vertices 變量非常大，你廣播的時(shí)候存在OOM 的風(fēng)險(xiǎn)，但是如果你使用了SparkSQL的話，Spark 就會(huì)根據(jù)實(shí)際的情況，幫你自動(dòng)優(yōu)化。

優(yōu)化點(diǎn) 增量?jī)?yōu)化

我們看到我們每次都是全量的圖，其實(shí)我們可以將我們的OneID 表加載進(jìn)來(lái)，然后將我們的增量數(shù)據(jù)和已有的圖數(shù)據(jù)進(jìn)行合并，然后再去生成圖

val?veritx?=?ye_veritx.union(to_veritx)
val?edges?=?ye_edges.union(to_edges)

val?graph?=?Graph(veritx,?edges)

總結(jié)

1. ID Mapping 是OneID 的提前，OneID 是ID Mapping 的結(jié)果，所以要想做OneID必須先做ID Mapping;
2. OneID 是為了打通整個(gè)數(shù)據(jù)體系的數(shù)據(jù)，所以OneID 需要以服務(wù)的方式對(duì)外提供服務(wù)，在數(shù)倉(cāng)里面就是作為基礎(chǔ)表使用，對(duì)外的話我們就需要提供接口對(duì)外提供服務(wù)