Spark中ID發號器實現思路

spark作為乙個分布式處理框架，處理資料非常快。可是我也不知道作者基於什麼樣的設計哲學，限制了使用者對於一些資料的操作。例如你無法改變乙個rdd的內容。無法乙個split把資料集分割成兩份。無法獲得rdd的分割槽資訊，無法再map的時候知道自己處於哪個分割槽。這對於id分配來說實在是太難辦了。甚至想要乙個1對1的笛卡爾積都沒有。（如果有的話，我們可以計算待分配id資料的個數，然後生成乙個資料個數長度的id列表，然後分配給他們即可）

經過我好幾天的調查，嘗試了很多方法。

最容易想到的方法將資料tolocaliterator，然後再給每條分配id。再join回去，這個方法很慢。而且會把資料放在乙個節點上。代價太高。

網上給出的方法也主要是使用其他服務，例如zookeeper來做乙個全域性請求的發號器。

經過個人研究，實現了兩套發號器思路。

1.最快速，最簡單的方法適用於id不要求連續的情況，並且資料數目很少的情況。此時可以使用乙個hash演算法，根據資料的key來計算它的id。資料量少hash碰撞概率很小。畢竟id最主要的目的還是標識資料。這裡推薦使用murmurhash。scala**如下，這裡charsets根據自己實際上求hash的key的字符集而定我這裡是給url分配的，因此asc2的字符集就夠（字符集越小，hash碰撞概率也越小）：

import com.google.common.base.charsets
import com.google.common.hash.hashing
hashing.murmur3_128().hashstring(str, charsets.us_ascii).aslong.abs

2.是今天剛想到的乙個方法，思路是這樣：

利用spark的glom(),將rdd的每個分割槽轉化成array。這樣就可以統計每個分割槽的元素個數。然後利用個數計算每個分割槽的資料id起始下標，將其存到乙個map中，然後廣播。話不多說上**：

假設你的rdd存的是一行行資料，你此次分配的起始id為beginid是1。有兩種思路獲取帶id的part第一種注釋掉了，因為這種只能處理資料不同的情況，並且需要兩步操作也會多一些。

val beginid = 1
val newmaxid = beginid + allnoiddata.count()//獲取此次分配的最大id備用
//val allurlpartarrayold = allnoidurl.glom().map(e => 
// (if (e.length > 0) e.head else "empty", e)
//).persist(storagelevel.memory_and_disk_ser_2)
iterator((partid,iter))//變成了partid+這個part的資料。
}).persist(storagelevel.memory_and_disk_ser_2)
val keynummap = allurlpartarray.map(e => (e._1, e._2.length)).collectasmap()
val keybeginidmap = mutable.hashmap[int, long]()
keynummap.foreach(e => )
val keybeginidmapbroadcast = sc.broadcast(keybeginidmap)
val dataidrdd = allurlpartarray.flatmap(e => )
})

如果有感興趣的可以相互交流。

2018-11-14後記:

再一次表示自己孤陋寡聞了，zipwithindex()專門做這個事的。

Spark中ID發號器實現思路

發號器的設計

分布式發號器 Vesta

VS中ID號的格式ID ,IDS ,IDC

Spark中ID發號器實現思路

發號器的設計

分布式發號器 Vesta

VS中ID號的格式ID ,IDS ,IDC

相關推薦