提出需求

實時統計業務系統(web,APP之類)的訪問人數,即所謂UV,或者DAU指標.

這個需求怕是流計算最最最常見的需求了.

計算UV的關鍵點就在於去重,即同一個人訪問兩次是隻計一個UV的.在離線計算中統計UV比較容易想到的方法就是用group或distinct機制來去重.但是在實時計算場景,還用group就不太科學了,一個是全量資料的group是比較費時的,第二個是全量資料的group是很費記憶體和CPU的.特別是當用戶量巨大的時候,還要做到秒級更新就更難了.

總結起來,需求就是:海量使用者場景UV實時計算.

接受挑戰

不難發現,問題的主要難點就是去重.

Spark Streaming目前沒有給出內建方案(這個其實可以有),但是海量資料去重問題早就有解決辦法了.
所以Spark Streaming程式完全可以利用其他系統的現有方案解決去重問題,比如Redis.

Redis的海量去重計數方案

Bitmap方案

所謂的Bitmap就是用一個bit位來標記某個元素對應的Value,比如ID為2的使用者,就用第2個bit位來表示,然後用該位的值來表示該使用者是否訪問過.如果要計算UV,那就只要數一下有多少個1就行啦.

假設我們有40億使用者,使用Bitmap需要2^32個bit位,算下來也就500M左右.

你可能沒想到的是,Redis中最常用的資料結構string,就可以實現bitmap演算法.

Redis提供瞭如下命令

// 插入
setbit key offset value
//獲取
getbit key offset
//計數
BITCOUNT key [start] [end
 
]

這裡offset最大值就是2^32.
比如ID為2的使用者,可以setbit uv 2 1,來記錄.
最後要計算UV,就直接 BITCOUNT uv. 這步計數非常快,複雜度是O(1).

HyperLogLog方案

若要計算很多頁面的UV,用bitmap還是比較費空間的,N個頁面就得有N個500M.這時候HyperLogLog結構就是一個比較好的選擇.

Redis 在 2.8.9 版本添加了 HyperLogLog 結構。
Redis HyperLogLog 是用來做基數統計的演算法，HyperLogLog 的優點是，在輸入元素的數量或者體積非常非常大時，計算基數所需的空間總是固定 的、並且是很小的。
在 Redis 裡面，每個 HyperLogLog 鍵只需要花費 12 KB 記憶體，就可以計算接近 2^64 個不同元素的基 數。這和計算基數時，元素越多耗費記憶體就越多的集合形成鮮明對比。
但是，因為 HyperLogLog 只會根據輸入元素來計算基數，而不會儲存輸入元素本身，所以 HyperLogLog 不能像集合那樣，返回輸入的各個元素。

也就是說HyperLogLog是一種基數統計演算法,計算結果是近似值, 12 KB 記憶體就可以計算2^64 個不同元素的基數.

Redis命令如下:

redis 127.0.0.1:6379> PFADD runoobkey "redis"

1) (integer) 1

redis 127.0.0.1:6379> PFADD runoobkey "mongodb"

1) (integer) 1

redis 127.0.0.1:6379> PFADD runoobkey "mysql"

1) (integer) 1

redis 127.0.0.1:6379> PFCOUNT runoobkey

(integer) 3

程式碼實現

下面給出HyperLogLog方案的參考實現:

stream.foreachRDD { rdd =>
    //統計人數
    rdd.foreachPartition { partition =>
        //從分割槽所屬executor的redis執行緒池獲取一個連線.
        val redis = RedisUtil.getRedis
        partition.foreach { case (date, userId) =>
            //統計當前userId
            redis.pfadd(s"uv:$date", userId)
        }
        redis.close()
    }
}

關於Redis的連線,如果是用java或scala可以使用JedisPool,注意處理序列化即可.