問(wèn)題：
因?yàn)橘~號(hào)體系的特殊性，customer_id和fans_id是一對(duì)一的，而weixin_openid是微信那邊通過(guò)用戶的id跟用戶關(guān)注的公眾號(hào)id做hash之后的結(jié)果，所以現(xiàn)有賬戶體系下我們沒(méi)辦法知道某一個(gè)人跨店鋪買了哪些東西，而這個(gè)問(wèn)題在關(guān)聯(lián)規(guī)則挖掘以及找共現(xiàn)矩陣的時(shí)候就會(huì)出問(wèn)題，同一個(gè)人在其他店鋪里的購(gòu)買行為我們也沒(méi)辦法知道，雖然我們有我們自己的賬戶體系，但是現(xiàn)在很多用戶并不需要注冊(cè)我們的自己賬號(hào)就可以完成交易。
解決方案：
用session_id來(lái)粗略地定位到一個(gè)用戶，同一個(gè)sessionid對(duì)應(yīng)的幾個(gè)fansid認(rèn)為是同一個(gè)人，但是因?yàn)閟essionid會(huì)定期清除更新，所以同一個(gè)fansid的sessionid可能會(huì)出現(xiàn)多個(gè)，不同sessionid對(duì)應(yīng)的fansid的集合如果有交集，我們認(rèn)為這是同一個(gè)用戶，應(yīng)該把兩個(gè)集合合并。
具體步驟：
取出hive里存的nginxlog里的sessionid跟fansid都不為空的記錄，存進(jìn)一張新表里；
用mapreduce把hive里存的數(shù)據(jù)根據(jù)session_id來(lái)merge，同一個(gè)session_id下面的用戶認(rèn)為是一個(gè)人，處理結(jié)果類似這樣：
sessionid1: fansid1, fansid2,fansid3
sessionid2: fansid1, fansid12,fansid13
……
- 對(duì)上面的結(jié)果進(jìn)行拆分，結(jié)果類似這樣：
fansid1:fansid1,fansid2,fansid3
fansid2:fansid1,fansid2,fansid3
fansid3:fansid1,fansid2,fansid3
fansid1:fansid1, fansid12,fansid13
fansid12:fansid1, fansid12,fansid13
fansid13:fansid1, fansid12,fansid13
……
- 對(duì)相同的fansid做merge，結(jié)果類似這樣：
fansid1:fansid1,fansid2,fansid3,fansid12,fansid13
fansid2:fansid1,fansid2,fansid3
fansid3:fansid1,fansid2,fansid3
fansid12:fansid1, fansid12,fansid13
fansid13:fansid1, fansid12,fansid13
......
- 找存在交集的fansid，認(rèn)為是同一個(gè)用戶，針對(duì)每個(gè)fansid，需要找出他關(guān)聯(lián)的每個(gè)fansid，然后取出關(guān)聯(lián)的這個(gè)fansid關(guān)聯(lián)的fansid，然后取并集。舉例來(lái)說(shuō)，fansid1，首先取fansid2，然后取出fansid2關(guān)聯(lián)的fansid：fansid1,fansid2,fansid3，然后做并集，（fansid1,fansid2,fansid3,fansid12,fansid13）并（fansid1,fansid2,fansid3），然后繼續(xù)取fansid3.....
這里有一個(gè)問(wèn)題，就是需要存儲(chǔ)中間結(jié)果，用來(lái)取每個(gè)關(guān)聯(lián)fansid各自的fansid，原先的數(shù)據(jù)是存hdfs的，沒(méi)法做K/V的實(shí)時(shí)查詢，所以需要一個(gè)中間結(jié)果存儲(chǔ)介質(zhì)。我們的數(shù)據(jù)行數(shù)近1kw。有幾種存儲(chǔ)方案可以選擇，這里嘗試了兩種。
一種是把mapreduce計(jì)算出的中間結(jié)果— hdfs上的文件通過(guò)DistributedCache緩存到各個(gè)任務(wù)節(jié)點(diǎn)，DistributedCache會(huì)把hdfs上的文件拷到各自任務(wù)節(jié)點(diǎn)的作為一個(gè)臨時(shí)文件，然后我們?cè)谌蝿?wù)的setup階段把文件讀進(jìn)內(nèi)存里自己定義的數(shù)據(jù)結(jié)構(gòu)里，結(jié)果讀進(jìn)去就OOM了，不要說(shuō)后面還要基于這個(gè)數(shù)據(jù)做其他的操作了。
第二種是現(xiàn)在的方案，中間結(jié)果存hbase，rowkey用fansid，列存關(guān)聯(lián)的fansid，因?yàn)橥莌adoop家族成員，mapreduce和hbase的交互很方便，有很好用的API可以直接用。之前的mapreduce寫hdfs文件的方式改成直接寫進(jìn)hbase，后面mapreduce任務(wù)直接根據(jù)一定的切分規(guī)則各自讀取指定段的hbase里的數(shù)據(jù)進(jìn)行處理，處理過(guò)程中需要到hbase里根據(jù)fansid取關(guān)聯(lián)fansid做并集，所以有1kw行就需要訪問(wèn)至少1kw次hbase，這里的瓶頸就在訪問(wèn)hbase了。后面看了一些可以過(guò)濾的情況，訪問(wèn)次數(shù)被大幅減少了。
一個(gè)完整的例子（hive初始數(shù)據(jù)處理忽略）：
下面是fansid跟關(guān)聯(lián)的fansid的映射：
1:1,2,3
2:1,2,3
3:1,2,3
2:2,3,4
3:2,3,4
4:2,3,4
1:1,5
5:1,5
5:5,6
6:5,6
6:6,7
7:6,7
8:8,9,10
9:8,9,10
10:8,9,10
通過(guò)我們上面的處理，最后的映射關(guān)系應(yīng)該是：
1 3,2,1,5,4,6,7
10 10,9,8
2 3,2,1,4,5,6,7
3 3,2,1,4,5,6,7
4 3,2,4,1,5,6,7
5 1,6,5,3,2,7,4
6 7,6,5,1,3,2,4
7 7,6,5,1,3,2,4
8 10,9,8
9 10,9,8
具體流程備忘：
- hive處理nginx_log表。
insert overwrite table sessionid_releated_fansid partition (par='2015') select distinct kdtsessionid, fans_id from nginx_log where kdtsessionid <> 'none' and fans_id <> 'none';
- map reduce任務(wù)處理上面的結(jié)果，得到fansid與關(guān)聯(lián)的fansid的映射
hadoop fs -rm -r /user/hive/warehouse/st.db/sessionid_releated_fansid/res=2015
hadoop jar associationgoods-0.0.1-SNAPSHOT-jar-with-dependencies.jar com.kdt.datapreprocess.GenerateRelatedFansIdJob /user/hive/warehouse/st.db/sessionid_releated_fansid/par=2015 /user/hive/warehouse/st.db/sessionid_releated_fansid/res=2015
注：行數(shù)應(yīng)該等于上面的文件的行數(shù)
- 兩個(gè)map reduce任務(wù)順序執(zhí)行，第一個(gè)merge上面的fansid，第二個(gè)查找每個(gè)fancied關(guān)聯(lián)的所有fansid，得到最終fansid與關(guān)聯(lián)的fansid的映射
truncate 'related_fansid_intermediate_result'
hadoop fs -rm -r /user/hive/warehouse/st.db/sessionid_releated_fansid/final_res
hadoop jar associationgoods-0.0.1-SNAPSHOT-jar-with-dependencies.jar com.kdt.datapreprocess.MergeFansIdJob /user/hive/warehouse/st.db/sessionid_releated_fansid/res=2015 /user/hive/warehouse/st.db/sessionid_releated_fansid/final_res
注：hbase總行數(shù)計(jì)算：
hbase org.apache.hadoop.hbase.mapreduce.RowCounter 表名
