程式扎記

參考至自己動手寫網路爬蟲 ISBN:9866143309
前言 :
分散式儲存是網路爬蟲中一個重要的課題, 抓取下來的 URL 要存在分散式環境中. 在雲端運算熱潮風起雲湧的今天, 儲存雲的概念也被炒得沸沸揚楊. 因此如何在分散式網路環境中儲存資料, 也是分散式爬蟲的重要課題之一.

從 Relation_DB 到 key/value 儲存 :
幾年前 key/value 這個詞還是和 hash 表關聯在一起的. 而現在程式師看到 key/value 這個詞時, 馬上聯想到的就是 BigTable, SimpleDB 和雲端運算. 當下 key/value 儲存 (或者叫 key/value Database, 雲端儲存等) 是個非常時髦的詞彙, 越來越多的開發人員 (特別是網際網路企業) 開始關心和嘗試 key/value 的儲存形式. key/value 形式的儲存並不是憑空想像出來的, 有兩個原因導致了 key/value 儲存方式的崛起.

* 大規模的網際網路應用
對於 Google 和 eBay 這樣的網際網路企業, 每時每刻都有無數的使用者在使用它們提供的網際網路服務, 這些服務帶來的就是大量的資料傳輸量. 同一時間會並發出成千上萬的連接對資料庫的操作. 在這種情況下單台的伺服器或者幾台伺服器遠遠不能滿足這些資料處理的需求, 簡單的升級伺服器的效能的方式也不行, 所以唯一可以採取的辦法就是使用叢集了. 使用叢集的方法有很多種, 但大致分為兩類 : 一類仍然採用關聯式資料庫 (RDBMS), 然後在透過對資料庫的垂直和水平切割將整個資料庫部署到一個叢集上, 這種方式優點在於可以採用 RDBMS 這種熟悉的技術, 但缺點在於它是針對特定應用. 由於應用的不同, 切割的方法是不一樣的.

還有一類就是 Google 所採用的方法, 拋棄 RDBMS, 採用 key/value 形式儲存. 這樣可以極大的增強系統的可擴充性 (scalability), 如果要處理的資料量持續增大, 多加機器就可以了. 事實上 key/value 的儲存就是由於 BigTable 等相關論文的發表進入人們的視野.

* 雲端儲存
如果說上一個問題還有可以替代的解決方案 (切割資料庫) 的話, 那麼對於雲端儲存來說, 也許 key/value 的儲存就是唯一的解決方案. 雲端儲存簡單說就是建構一個大型的儲存平台給別人用, 這也就表示在這上面執行的應用其實是不可控制的. 如果其中某個客戶的應用隨著使用者的增長而不斷擴張時, 雲端儲存供應商是沒有辦法透過資料庫的切割來達到擴充的, 因為這個資料是客戶的, 供應商不了解這個資料自然就沒法做出切割. 在這種情況 key/value 的儲存就是唯一的選擇. 因為這種條件下的可擴充性必須是自動完成的, 不能有人工干預. 這也是為什麼幾乎目前所有雲端都是 key/value 形式, 例如 Amazon 的 simpleDB, 底層實現的就是 key/value, 還有 Google 的 GoogleAppEngine, 採用的是 BigTable 的儲存形式.

key/value 儲存與 RDBMS 相比, 一個很大的區別就是它沒有模式的概念. 在 RDBMS 中, 模式所代表的其實就是對資料的約束, 包括資料間的關係 (relationship) 和資料的完整性 (integrity), 例如 RDBMS 中對於某個資料屬性會要求它的資料類型是確定的 (整數或字串 etc), 資料的範圍也是確定的, 而這些在 key/value 儲存中都沒有. 在 key/value 儲存中, 對於某個 key, value 可以是任何類型的資料.

在所有 RDBMS 中, 都是採用 SQL 語言對資料進行存取. 一方面 SQL 對於資料的查詢非常強大 ; 另一方面由於所有的 RDBMS 都支援 SQL 查詢, 所以可攜性很強. 而在 key/value 儲存中, 對於資料的操作使用都是自訂的一些 API, 而且支援的查詢也相對簡單. 正如前面提到的 , key/value 儲存最大特點就是它的可擴充性. 所謂的擴充性其實包括兩方面內容. 一方面是指 key/value 儲存可以支援極大的資料儲存. 它的分散式架構決定了只要有更多的機器, 就能保證儲存更多的資料. 另一方面是指它可以支援數量很多的同時查詢. 對於 RDBMS, 一般幾百個同時查詢就可以讓它很吃力了, 而一個 key/value 儲存可以很輕鬆的支援上千個同時查詢.

但沒有十全十美的方案, key/value 儲存有幾個主要缺陷 :

- 由於 key/value 儲存沒有 schema, 所以它是不提供資料間的關係和資料的完整性, 所有的這些東西都落到了應用程式一端, 也就是開發人員. 無疑加重開發人員負擔.

- 在 RDBMS 中, 需要設定各表間的關係, 這其實是一個資料建模的過程 (data modeling process). 當資料建模完成, 這個資料庫對應用程式就是獨立的, 這表示其它程式可以在不改變資料模型前提下使用相同的資料集. 但在 key/value 儲存中, 由於沒有一個這樣的資料模型, 不同的應用程式需要重複進行這個過程.

- key/value 儲存最大的一個缺點在於它的介面是不友善的. 這阻礙了開發人員可以快速且順利的應用它. 當然現在有種做法就是在 key/value 儲存上再加上一個類似 SQL 敘述的抽象介面層, 從而使得開發人員可以用他們熟悉的方式 (SQL) 來操作 key/value 儲存. 但由於 RDBMS 和 key/value 儲存的底層實現有著很大的不同, 這種抽象介面幫助有限.

Consistent Hash 演算法 :
分散式儲存常常會涉及負載平衡的問題, 由於有多個儲存媒體, 分布在不同的節點上. 因此當一個物件被儲存時, 他究竟應該儲存在哪個儲存媒體上 (儲存媒體可以是資料庫, Berkeley DB 等, 甚至可以是記憶體資料結構)? 這就是負載平衡的問題, 如下圖所示 :

面對雲端運算, 如何很好的分布儲存資料, 是一個非常重要的話題. 在分散式爬蟲中, 抓取的頁面非常多 (通常是數十億級別), 因此分散式儲存就非常有意義. 那麼如果抓取下來一個頁面, 究竟要存放到哪個資料庫中呢? 這就涉及負載均衡的問題.

考慮如果你有 N 個資料儲存的伺服器, 那麼怎麼將一個物件映射到 N 個伺服器上? 你可能會採用類似下面的通用方法計算物件的 hash 值, 然後均勻的映射到 N 個伺服器 :

hash(object) % N

一切都執行正常, 但是要考慮以下兩種狀況 :

(1) 一個伺服器 m 掛掉 (在實際應用中必須要考慮這種狀況), 則所有映射到伺服器 m 的物件都會失效. 並且需要把伺服器 m 移除, 這時伺服器為 N-1 台, 映射公式變成了 hash(object) % (N-1)
(2) 由於存取加重, 需要增加伺服器, 這時候變成了 N+1 台伺服器, 映射公式變成 hash(object) % (N+1)

在上面兩種狀況, 突然之間幾乎所有的伺服器都失效了. 對於伺服器而言, 這是一場災難. 再來考慮第三個問題, 由於硬體能力越來越強, 你可能想讓後面增加的節點多做點事, 顯然上面的 hash 演算法也做不到. 有什麼方法可以改變這個狀況呢? 這就要用到 Consistent Hashing 演算法.

hash 的演算法一個衡量指標是單調性 (Monotonicity) 定義如下 :

單調性是指如果已經有一些內容透過雜湊分配到相應的緩衝中, 而又有新的緩衝加到系統中. 雜湊的結果應該能保證原有已分配的內容可以映射到新的緩衝中, 而不會被映射到舊的緩衝集合中的其他緩衝區.

上面的簡單 hash 演算法 hash(object)%N 難以滿足單調性的要求.

Consistent Hashing 是一種 hash 演算法, 簡單的說在移除/新增一個伺服器時, 它能盡可能小地改變已存在的 key 映射關係, 盡可能的滿足單調性的要求. 下面就按照 5 個步驟簡單說明 Consistent Hashing 演算法的基本原理.

* 步驟一 : 環形 hash 空間
考慮通常的 hash 演算法都是將 value 映射到一個 32 位的 key 值, 即 0~2^(32-1) 的數值空間. 我們可以將這個空間想像成一個首尾相接的圓環.

* 步驟二 : 把物件映射到 hash 空間
接下來考慮 4 個物件 object1~object4, 透過 hash 函數計算出 hash 值 key 在環上的分布 :

hash(object1) = key1
...
hash(object4) = key4

* 步驟三 : 將伺服器映射到 hash 空間
Consistent Hashing 的基本思想就是將物件和伺服器都映射到同一個 hash 數值空間, 並且使用相同的 hash 演算法. 假設目前有 A, B 和 C 三台伺服器, 那麼其映射結果如下圖所示, 他們在 hash 空間中, 以對應的 hash 值排列 :

hash(伺服器 A) = key A;
...
hash(伺服器 C) = key C;

* 步驟四 : 把物件映射到伺服器
現在 cache 和物件都已經透過同一個 hash 演算法映射到 hash 數值空間.接下來要考慮的就是如何將物件映射到 cache 上面. 在這個環形空間, 如果沿著順時鐘方向從物件的 key 值出發, 直到遇見第一個伺服器, 那麼就將該物件儲存在這個伺服器上, 因為物件和伺服器是固定的, 因此這個伺服器必然是唯一與確定的. 這樣就找到物件與伺服器的對應方法了. 由上圖可以知道物件 object1 被儲存在伺服器 A 上, object2 和 object3 對應到伺服器 C, object 4 對應到伺服器 B.

* 步驟五 : 考察伺服器的變動
前面講過透過 hash 演算法後求餘帶來最大問題就在於不能滿足單調性, 當伺服器有所變動, 伺服器會失效, 進而對服務造成中斷. 現在我們就透過實際範例來說明 Consistent Hashing 演算法.
(1) 移除伺服器
考慮假設伺服器 B 掛掉, 根據前面說明的映射方法, 這時受到影響的只有在 cache B 上面的物件, 只要從伺服器 B 逆時針檢查就可以知道那些物件受到影響. 因此這裡僅需要變動物件 object4 , 將其重新映射到伺服器 C 上即可, 如下圖 :

(2) 增加伺服器
在考慮增加一台新的伺服器 D 的情況, 假設在這個環形 hash 空間中, 伺服器 D 被映射到物件 object2 與 object3 之間, 這時受影響的僅是那些沿 cache D 逆時針檢查直到下一個伺服器之間的物件, 將這些物件重新映射到伺服器 D 上即可. 因此這裡僅需要變動物件 object2, 並將其重新映射到伺服器 D 上.

考量 hash 演算法的另一個指標是平衡性 (Balance), 定義如下 :

平衡性是指雜湊的結果能夠盡可能分布到所有緩衝中, 這樣可以使所有的緩衝空間都得到利用.

hash 演算法並不能保證絕對的平衡, 如果伺服器少, 物件並不能被均勻地映射到伺服器上, 例如上面的例子, 僅部署伺服器 A 和伺服器 C 的情況下, 在 4 個物件中, 伺服器 A 僅儲存物件 object1, 而伺服器 c 儲存了剩下的 3 個物件, 分布是很不平衡的. 為了解決這種情況, Consistent Hashing 引入了 "虛擬節點" 的概念, 他可以如下定義 :

"虛擬節點" (virtual node) 是實際節點在 hash 空間的複製品 (replica), 一個實際的節點對應若干個 "虛擬節點" , 這個對應個數也稱為 "複製個數" , "虛擬節點" 在 hash 空間中以 hash 值排列.

仍然僅部署伺服器 A 和伺服器 C 的情況為例, 在上圖我們看到伺服器分布並不均勻. 現在我們引入虛擬節點, 並設定 "複製個數" 為 2, 這表示一共會存在 4 個 "虛擬節點", 伺服器A1 和伺服器A2 代表伺服器 A ; 伺服器 C1 和伺服器 C2 代表伺服器 C, 假設一種比較理想的情況如下所示 :

此時物件到 "虛擬節點" 的映射關係為 : object1-> 伺服器 A2; object2->伺服器 A1; object3->伺服器 C1; object4->伺服器C2.

因此物件 object1 和 object2 都被映射到伺服器A上, 而object3, object4 則被映射到伺服器C 上, 平衡性上有了很大的改善. 另外引入 "虛擬節點" 後, 映射關係就從 {物件->節點} 轉換到 {物件->虛擬節點}. 查詢物件所在 cache 時的映射關係如下圖 :

Consistent Hash 範例實作 :
底下範例代碼為 Consistent Hash 的簡單實作, 並示範添加 Cache, 移除 Cache 與 Object 的添加.
- 代碼實作

- ConsistentHash.java : Consistent Hash 的簡化實作

view plain copy to clipboard print ?

package test.consistenthash;



import java.util.Iterator;

import java.util.Random;

import java.util.SortedMap;

import java.util.TreeMap;



public class ConsistentHash {

    private TreeMap roundMap = null;

    private int replic=1;

    private Random rdm;



    public ConsistentHash(int replic)

    {

        roundMap = new TreeMap();

        rdm = new Random();

        this.replic = replic;

    }



    public boolean addCache(Cache c)

    {

        if(replic>1)

        {

            for(int i=0; i

            {

                int hash = Math.abs((c.toString()+(i*1000+rdm.nextInt(1000))).hashCode());

                System.out.printf("\t[ConsistentHash] Cache(%s) in hash=%d...\n", c.name, hash);

                c.hashSet.add(hash);

                roundMap.put(hash, c);

            }

        }

        else

        {

            c.hashSet.add(c.hashCode());

            roundMap.put(c.hashCode(), c);

        }

        return true;

    }



    protected void _migrateObj(int h, Cache c)

    {

        SortedMap sm = roundMap.tailMap(h);

        if(sm.isEmpty()) sm = roundMap.tailMap(0);

        if(!sm.isEmpty())

        {

            Cache tmp = sm.get(sm.firstKey());

            Iterator kitr = tmp.keyIter();

            while(kitr.hasNext())

            {

                int k = kitr.next();

                if(k

                {

                    c.addObject(k, tmp.get(k));

                    kitr.remove();

                }

            }

        }

    }



    protected boolean _migrateObjInRemoveCache(Cache c)

    {

        Iterator kitr = c.keyIter();

        while(kitr.hasNext())

        {

            addDObject(c.get(kitr.next()));

        }

        return false;

    }



    public boolean addNewCache(Cache c)

    {

        int hash = c.hashCode();

        if(replic>1)

        {

            for(int i=0; i

            {

                hash = Math.abs((c.toString()+(i*1000+rdm.nextInt(1000))).hashCode());

                System.out.printf("\t[ConsistentHash] Add new Cache(%s) in hash=%d...\n", c.name, hash);

                if(!roundMap.isEmpty())

                {

                    if(roundMap.containsKey(hash)) continue; /*該 hash 已存在*/

                    _migrateObj(hash, c);

                }

                c.hashSet.add(hash);

                roundMap.put(hash, c);

            }

        }

        else

        {



            if(!roundMap.isEmpty())

            {

                if(roundMap.containsKey(hash)) return false; /*該 hash 已存在*/

                _migrateObj(hash, c);

            }

            c.hashSet.add(hash);

            roundMap.put(hash, c);

        }

        return true;

    }



    public boolean removeCache(Cache c)

    {

        Iterator hashItr = c.hashSet.iterator();

        while(hashItr.hasNext())

        {

            int h = hashItr.next();

            System.out.printf("\t[ConsistentHash] Remove Cache(%s) with hash=%d...\n", c.name, h);

            roundMap.remove(h);

        }

        _migrateObjInRemoveCache(c);

        return true;

    }



    public boolean addDObject(DistributeObj obj)

    {

        System.out.printf("\t[ConsistentHash] Add Obj(%s) hash=%d...\n", obj.name, obj.hashCode());

        int hash = obj.hashCode();

        if(roundMap.containsKey(hash))

        {

            Cache c = roundMap.get(hash);

            System.out.printf("\t[ConsistentHash] Match in Cache(%s)...\n", c.name);

            c.addObject(hash, obj);

            return true;

        }

        else

        {

            SortedMap sm = roundMap.tailMap(hash);

            if(sm.isEmpty()) sm = roundMap.tailMap(0);

            if(!sm.isEmpty())

            {

                int fkey = sm.firstKey();

                Cache c = sm.get(fkey);

                System.out.printf("\t[ConsistentHash] Match in Cache(%s) with hash=%d...\n", c.name,  fkey);

                c.addObject(hash, obj);

                return true;

            }

            System.out.printf("\t[ConsistentHash] No suitable Cache!\n");

            return false;

        }

    }



    public boolean removeObject(DistributeObj obj)

    {

        int hash = obj.hashCode();

        System.out.printf("\t[ConsistentHash] Remove Obj(%s) hash=%d...\n", obj.name, obj.hashCode());

        if(roundMap.containsKey(hash))

        {

            roundMap.get(hash).remove(hash);

            return true;

        }

        else

        {

            SortedMap sm = roundMap.tailMap(hash);

            if(sm.isEmpty()) sm = roundMap.tailMap(0);

            if(!sm.isEmpty())

            {

                int fkey = sm.firstKey();

                if(sm.get(fkey).remove(hash)!=null) return true;

            }

            return false;

        }

    }



    public int size(){return roundMap.size();}



    public static void main(String args[])

    {

        ConsistentHash consistentHash = new ConsistentHash(5);

        Cache c1 = new Cache("cache1");

        Cache c2 = new Cache("cache2");

        Cache c3 = new Cache("cache3");

        consistentHash.addCache(c1);

        consistentHash.addCache(c2);

        System.out.printf("\t[Test] ConsistentHash size=%d...\n", consistentHash.size());

        DistributeObj obj1 = new DistributeObj("obj1");

        DistributeObj obj2 = new DistributeObj("obj2");

        DistributeObj obj3 = new DistributeObj("obj3");

        DistributeObj obj4 = new DistributeObj("obj4");

        DistributeObj obj5 = new DistributeObj("obj5");

        consistentHash.addDObject(obj1);

        consistentHash.addDObject(obj2);

        consistentHash.addDObject(obj3);

        consistentHash.addDObject(obj4);

        System.out.printf("\t[Test] cache1 has %d object...\n", c1.size());

        System.out.printf("\t[Test] cache2 has %d object...\n", c2.size());

        System.out.printf("\t[Test] Remove obj(%s)...\n", obj2.name);

        consistentHash.removeObject(obj2);

        System.out.printf("\t[Test] cache1 has %d object...\n", c1.size());

        System.out.printf("\t[Test] cache2 has %d object...\n", c2.size());

        System.out.printf("\t[Test] Add new Cache(%s)...\n", c3.name);

        consistentHash.addNewCache(c3);

        System.out.printf("\t[Test] cache1 has %d object...\n", c1.size());

        System.out.printf("\t[Test] cache2 has %d object...\n", c2.size());

        System.out.printf("\t[Test] cache3 has %d object...\n", c3.size());

        System.out.printf("\t[Test] Add obj(%s) with hash=%d...\n", obj5.name, obj5.hashCode());

        consistentHash.addDObject(obj5);

        System.out.printf("\t[Test] cache1 has %d object...\n", c1.size());

        System.out.printf("\t[Test] cache2 has %d object...\n", c2.size());

        System.out.printf("\t[Test] cache3 has %d object...\n", c3.size());

        System.out.printf("\t[Test] Remove Cache(%s)...\n", c3.name);

        consistentHash.removeCache(c3);

        System.out.printf("\t[Test] cache1 has %d object...\n", c1.size());

        System.out.printf("\t[Test] cache2 has %d object...\n", c2.size());

    }

}

- Cache.java : 存放分布式儲存 Object 的 Cache

view plain copy to clipboard print ?

package test.consistenthash;



import java.util.HashSet;

import java.util.Iterator;

import java.util.TreeMap;



public class Cache {

    public HashSet hashSet = new HashSet();

    private TreeMap objMap;

    public String name = null;

    public Cache(String n)

    {

        objMap = new TreeMap();

        name = n;

    }



    public DistributeObj get(int h){return objMap.get(h);}

    public Iterator keyIter(){return objMap.keySet().iterator();}

    public void addObject(int h, DistributeObj obj) {objMap.put(h, obj);}

    public DistributeObj remove(int h){return objMap.remove(Integer.valueOf(h));}

    public int size(){return objMap.size();}

}

- DistributeObj.java : 模擬分布式 Object

view plain copy to clipboard print ?

package test.consistenthash;



public class DistributeObj {

    public String name = null;



    public DistributeObj(String n){this.name = n;}

    public DistributeObj(){name = this.getClass().getName();}

}

- 執行結果 :

[ConsistentHash] Cache(cache1) in hash=787955378...
[ConsistentHash] Cache(cache1) in hash=1343187047...
...(略)...
[ConsistentHash] Cache(cache2) in hash=385954449...
[Test] ConsistentHash size=10... # 兩個 Cache 乘 replic=5, 共有 10 個 Cache hash
[ConsistentHash] Add Obj(obj1) hash=758885650... # 開始添加分布式存放的 Object
[ConsistentHash] Match in Cache(cache1) with hash=787955378...
[ConsistentHash] Add Obj(obj2) hash=563152583...
[ConsistentHash] Match in Cache(cache1) with hash=787955378...
[ConsistentHash] Add Obj(obj3) hash=945030152...
[ConsistentHash] Match in Cache(cache1) with hash=1343187047...
[ConsistentHash] Add Obj(obj4) hash=332711452...
[ConsistentHash] Match in Cache(cache2) with hash=385954449...
[Test] cache1 has 3 object... # 共放入 4 個 Object, 分布到 2 個 Cache
[Test] cache2 has 1 object...
[Test] Remove obj(obj2)... # 移除 Object2
[ConsistentHash] Remove Obj(obj2) hash=563152583...
[Test] cache1 has 2 object...
[Test] cache2 has 1 object...
[Test] Add new Cache(cache3)... # 添加 Cache, 會造成某些 Object 的 migrate...
[ConsistentHash] Add new Cache(cache3) in hash=2135832963...
[ConsistentHash] Add new Cache(cache3) in hash=1786186606...
[ConsistentHash] Add new Cache(cache3) in hash=1786225172...
[ConsistentHash] Add new Cache(cache3) in hash=1786254097...
[ConsistentHash] Add new Cache(cache3) in hash=1786282830...
[Test] cache1 has 2 object...
[Test] cache2 has 0 object...
[Test] cache3 has 1 object... # Cache 2 上的一個 Object migrate 到 Cache3 上.
[Test] Add obj(obj5) with hash=1223001434...
[ConsistentHash] Add Obj(obj5) hash=1223001434...
[ConsistentHash] Match in Cache(cache1) with hash=1343187047...
[Test] cache1 has 3 object...
[Test] cache2 has 0 object...
[Test] cache3 has 1 object...
[Test] Remove Cache(cache3)... # 移除 Cache3, 則 Cache3 上的 Objects 會 migrate 到其他 Cache 上
[ConsistentHash] Remove Cache(cache3) with hash=1786254097... # 先移除 Round Map 上 Cache3 的點.
[ConsistentHash] Remove Cache(cache3) with hash=1786225172...
[ConsistentHash] Remove Cache(cache3) with hash=1786282830...
[ConsistentHash] Remove Cache(cache3) with hash=1786186606...
[ConsistentHash] Remove Cache(cache3) with hash=2135832963...
[ConsistentHash] Add Obj(obj4) hash=332711452... #將 Cache3 上的 Object 移到 Cache2 上.
[ConsistentHash] Match in Cache(cache2) with hash=385954449...
[Test] cache1 has 3 object...
[Test] cache2 has 1 object...

This message was edited 20 times. Last update was at 09/12/2011 12:46:13

程式扎記

標籤

2011年12月9日星期五

[ Java Crawler ] 分散式爬蟲 : 分散式儲存

沒有留言:

張貼留言

[Git 常見問題] error: The following untracked working tree files would be overwritten by merge

檢舉濫用情形

學習筆記

標籤

2011年12月9日 星期五