你真的了解實(shí)時(shí)計(jì)算嗎？
??
實(shí)時(shí)計(jì)算是什么？
請(qǐng)看下面的圖：
我們以熱賣產(chǎn)品的統(tǒng)計(jì)為例，看下傳統(tǒng)的計(jì)算手段：
- 將用戶行為、log等信息清洗后保存在數(shù)據(jù)庫(kù)中.
- 將訂單信息保存在數(shù)據(jù)庫(kù)中.
- 利用觸發(fā)器或者協(xié)程等方式建立本地索引，或者遠(yuǎn)程的獨(dú)立索引.
- join訂單信息、訂單明細(xì)、用戶信息、商品信息等等表，聚合統(tǒng)計(jì)20分鐘內(nèi)熱賣產(chǎn)品，并返回top-10.
- web或app展示.
這是一個(gè)假想的場(chǎng)景，但假設(shè)你具有處理類似場(chǎng)景的經(jīng)驗(yàn)，應(yīng)該會(huì)體會(huì)到這樣一些問題和難處：
- 水平擴(kuò)展問題（scale-out）
顯然，如果是一個(gè)具有一定規(guī)模的電子商務(wù)網(wǎng)站，數(shù)據(jù)量都是很大的。而交易信息因?yàn)樯婕笆聞?wù)，所以很難直接舍棄關(guān)系型數(shù)據(jù)庫(kù)的事務(wù)能力，遷移到具有更好的scale-out能力的NoSQL數(shù)據(jù)庫(kù)中。
那么，一般都會(huì)做sharding。歷史數(shù)據(jù)還好說，我們可以按日期來歸檔，并可以通過批處理式的離線計(jì)算，將結(jié)果緩存起來。
但是，這里的要求是20分鐘內(nèi)，這很難。 - 性能問題
這個(gè)問題，和scale-out是一致的，假設(shè)我們做了sharding，因?yàn)楸矸稚⒃诟鱾€(gè)節(jié)點(diǎn)中，所以我們需要多次入庫(kù)，并在業(yè)務(wù)層做聚合計(jì)算。
問題是，20分鐘的時(shí)間要求，我們需要入庫(kù)多少次呢？
10分鐘呢？
5分鐘呢？
實(shí)時(shí)呢？
而且，業(yè)務(wù)層也同樣面臨著單點(diǎn)計(jì)算能力的局限，需要水平擴(kuò)展，那么還需要考慮一致性的問題。
所以，到這里一切都顯得很復(fù)雜。 - 業(yè)務(wù)擴(kuò)展問題
假設(shè)我們不僅僅要處理熱賣商品的統(tǒng)計(jì)，還要統(tǒng)計(jì)廣告點(diǎn)擊、或者迅速根據(jù)用戶的訪問行為判斷用戶特征以調(diào)整其所見的信息，更加符合用戶的潛在需求等，那么業(yè)務(wù)層將會(huì)更加復(fù)雜。
也許你有更好的辦法，但實(shí)際上，我們需要的是一種新的認(rèn)知：
這個(gè)世界發(fā)生的事，是實(shí)時(shí)的。
所以我們需要一種實(shí)時(shí)計(jì)算的模型，而不是批處理模型。
我們需要的這種模型，必須能夠處理很大的數(shù)據(jù)，所以要有很好的scale-out能力，***是，我們都不需要考慮太多一致性、復(fù)制的問題。
那么，這種計(jì)算模型就是實(shí)時(shí)計(jì)算模型，也可以認(rèn)為是流式計(jì)算模型。
現(xiàn)在假設(shè)我們有了這樣的模型，我們就可以愉快地設(shè)計(jì)新的業(yè)務(wù)場(chǎng)景：
- 轉(zhuǎn)發(fā)最多的微博是什么？
- 最熱賣的商品有哪些？
- 大家都在搜索的熱點(diǎn)是什么？
- 我們哪個(gè)廣告，在哪個(gè)位置，被點(diǎn)擊最多？
或者說，我們可以問：
這個(gè)世界，在發(fā)生什么？
最熱的微博話題是什么？
我們以一個(gè)簡(jiǎn)單的滑動(dòng)窗口計(jì)數(shù)的問題，來揭開所謂實(shí)時(shí)計(jì)算的神秘面紗。
假設(shè)，我們的業(yè)務(wù)要求是：
統(tǒng)計(jì)20分鐘內(nèi)最熱的10個(gè)微博話題。
解決這個(gè)問題，我們需要考慮：
- 數(shù)據(jù)源
這里，假設(shè)我們的數(shù)據(jù)，來自微博長(zhǎng)連接推送的話題。 - 問題建模
我們認(rèn)為的話題是#號(hào)擴(kuò)起來的話題，最熱的話題是此話題出現(xiàn)的次數(shù)比其它話題都要多。
比如：@foreach_break : 你好,#世界#,我愛你，#微博#。
“世界”和“微博”就是話題。 - 計(jì)算引擎
我們采用storm。 - 定義時(shí)間
如何定義時(shí)間？
時(shí)間的定義是一件很難的事情，取決于所需的精度是多少。
根據(jù)實(shí)際，我們一般采用tick來表示時(shí)刻這一概念。
在storm的基礎(chǔ)設(shè)施中，executor啟動(dòng)階段，采用了定時(shí)器來觸發(fā)“過了一段時(shí)間”這個(gè)事件。
如下所示：
(defn setup-ticks! [worker executor-data]
(let [storm-conf (:storm-conf executor-data)
tick-time-secs (storm-conf TOPOLOGY-TICK-TUPLE-FREQ-SECS)
receive-queue (:receive-queue executor-data)
context (:worker-context executor-data)]
(when tick-time-secs
(if (or (system-id? (:component-id executor-data))
(and (= false (storm-conf TOPOLOGY-ENABLE-MESSAGE-TIMEOUTS))
(= :spout (:type executor-data))))
(log-message "Timeouts disabled for executor " (:component-id executor-data) ":" (:executor-id executor-data))
(schedule-recurring
(:user-timer worker)
tick-time-secs
tick-time-secs
(fn []
(disruptor/publish
receive-queue
[[nil (TupleImpl. context [tick-time-secs] Constants/SYSTEM_TASK_ID Constants/SYSTEM_TICK_STREAM_ID)]]
)))))))
之前的博文中，已經(jīng)詳細(xì)分析了這些基礎(chǔ)設(shè)施的關(guān)系，不理解的童鞋可以翻看前面的文章。
每隔一段時(shí)間，就會(huì)觸發(fā)這樣一個(gè)事件，當(dāng)流的下游的bolt收到一個(gè)這樣的事件時(shí)，就可以選擇是增量計(jì)數(shù)還是將結(jié)果聚合并發(fā)送到流中。
bolt如何判斷收到的tuple表示的是“tick”呢？
負(fù)責(zé)管理bolt的executor線程，從其訂閱的消息隊(duì)列消費(fèi)消息時(shí)，會(huì)調(diào)用到bolt的execute方法，那么，可以在execute中這樣判斷：
public static boolean isTick(Tuple tuple) {
return tuple != null
&& Constants.SYSTEM_COMPONENT_ID .equals(tuple.getSourceComponent())
&& Constants.SYSTEM_TICK_STREAM_ID.equals(tuple.getSourceStreamId());
}結(jié)合上面的setup-tick!的clojure代碼，我們可以知道SYSTEM_TICK_STREAM_ID在定時(shí)事件的回調(diào)中就以構(gòu)造函數(shù)的參數(shù)傳遞給了tuple，那么SYSTEM_COMPONENT_ID是如何來的呢？
可以看到，下面的代碼中，SYSTEM_TASK_ID同樣傳給了tuple：
;; 請(qǐng)注意SYSTEM_TASK_ID和SYSTEM_TICK_STREAM_ID
(TupleImpl. context [tick-time-secs] Constants/SYSTEM_TASK_ID Constants/SYSTEM_TICK_STREAM_ID)
然后利用下面的代碼，就可以得到SYSTEM_COMPONENT_ID：
public String getComponentId(int taskId) {
if(taskId==Constants.SYSTEM_TASK_ID) {
return Constants.SYSTEM_COMPONENT_ID;
} else {
return _taskToComponent.get(taskId);
}
}#p#
滑動(dòng)窗口
有了上面的基礎(chǔ)設(shè)施，我們還需要一些手段來完成“工程化”，將設(shè)想變?yōu)楝F(xiàn)實(shí)。
這里，我們看看Michael G. Noll的滑動(dòng)窗口設(shè)計(jì)。
注：圖片來自http://www.michael-noll.com/blog/2013/01/18/implementing-real-time-trending-topics-in-storm/
Topology
String spoutId = "wordGenerator";
String counterId = "counter";
String intermediateRankerId = "intermediateRanker";
String totalRankerId = "finalRanker";
// 這里，假設(shè)TestWordSpout就是我們發(fā)送話題tuple的源
builder.setSpout(spoutId, new TestWordSpout(), 5);
// RollingCountBolt的時(shí)間窗口為9秒鐘，每3秒發(fā)送一次統(tǒng)計(jì)結(jié)果到下游
builder.setBolt(counterId, new RollingCountBolt(9, 3), 4).fieldsGrouping(spoutId, new Fields("word"));
// IntermediateRankingsBolt，將完成部分聚合，統(tǒng)計(jì)出top-n的話題
builder.setBolt(intermediateRankerId, new IntermediateRankingsBolt(TOP_N), 4).fieldsGrouping(counterId, new Fields(
"obj"));
// TotalRankingsBolt， 將完成完整聚合，統(tǒng)計(jì)出top-n的話題
builder.setBolt(totalRankerId, new TotalRankingsBolt(TOP_N)).globalGrouping(intermediateRankerId);
上面的topology設(shè)計(jì)如下：
注：圖片來自http://www.michael-noll.com/blog/2013/01/18/implementing-real-time-trending-topics-in-storm/
將聚合計(jì)算與時(shí)間結(jié)合起來
前文，我們敘述了tick事件，回調(diào)中會(huì)觸發(fā)bolt的execute方法，那可以這么做：
RollingCountBolt:
@Override
public void execute(Tuple tuple) {
if (TupleUtils.isTick(tuple)) {
LOG.debug("Received tick tuple, triggering emit of current window counts");
// tick來了，將時(shí)間窗口內(nèi)的統(tǒng)計(jì)結(jié)果發(fā)送，并讓窗口滾動(dòng)
emitCurrentWindowCounts();
}
else {
// 常規(guī)tuple，對(duì)話題計(jì)數(shù)即可
countObjAndAck(tuple);
}
}
// obj即為話題，增加一個(gè)計(jì)數(shù) count++
// 注意，這里的速度基本取決于流的速度，可能每秒百萬，也可能每秒幾十.
// 內(nèi)存不足？ bolt可以scale-out.
private void countObjAndAck(Tuple tuple) {
Object obj = tuple.getValue(0);
counter.incrementCount(obj);
collector.ack(tuple);
}
// 將統(tǒng)計(jì)結(jié)果發(fā)送到下游
private void emitCurrentWindowCounts() {
Map<Object, Long> counts = counter.getCountsThenAdvanceWindow();
int actualWindowLengthInSeconds = lastModifiedTracker.secondsSinceOldestModification();
lastModifiedTracker.markAsModified();
if (actualWindowLengthInSeconds != windowLengthInSeconds) {
LOG.warn(String.format(WINDOW_LENGTH_WARNING_TEMPLATE, actualWindowLengthInSeconds, windowLengthInSeconds));
}
emit(counts, actualWindowLengthInSeconds);
}
上面的代碼可能有點(diǎn)抽象，看下這個(gè)圖就明白了，tick一到，窗口就滾動(dòng)：
注：圖片來自http://www.michael-noll.com/blog/2013/01/18/implementing-real-time-trending-topics-in-storm/
IntermediateRankingsBolt & TotalRankingsBolt：
public final void execute(Tuple tuple, BasicOutputCollector collector) {
if (TupleUtils.isTick(tuple)) {
getLogger().debug("Received tick tuple, triggering emit of current rankings");
// 將聚合并排序的結(jié)果發(fā)送到下游
emitRankings(collector);
}
else {
// 聚合并排序
updateRankingsWithTuple(tuple);
}
}其中，IntermediateRankingsBolt和TotalRankingsBolt的聚合排序方法略有不同：
IntermediateRankingsBolt的聚合排序方法：
// IntermediateRankingsBolt的聚合排序方法：
@Override
void updateRankingsWithTuple(Tuple tuple) {
// 這一步，將話題、話題出現(xiàn)的次數(shù)提取出來
Rankable rankable = RankableObjectWithFields.from(tuple);
// 這一步，將話題出現(xiàn)的次數(shù)進(jìn)行聚合，然后重排序所有話題
super.getRankings().updateWith(rankable);
}
TotalRankingsBolt的聚合排序方法：
// TotalRankingsBolt的聚合排序方法
@Override
void updateRankingsWithTuple(Tuple tuple) {
// 提出來自IntermediateRankingsBolt的中間結(jié)果
Rankings rankingsToBeMerged = (Rankings) tuple.getValue(0);
// 聚合并排序
super.getRankings().updateWith(rankingsToBeMerged);
// 去0，節(jié)約內(nèi)存
super.getRankings().pruneZeroCounts();
}
而重排序方法比較簡(jiǎn)單粗暴，因?yàn)橹磺笄癗個(gè)，N不會(huì)很大：
private void rerank() {
Collections.sort(rankedItems);
Collections.reverse(rankedItems);
}結(jié)語
下圖可能就是我們想要的結(jié)果，我們完成了t0 - t1時(shí)刻之間的熱點(diǎn)話題統(tǒng)計(jì)，其中的foreach_break僅僅是為了防盜版 : ].
文中對(duì)滑動(dòng)窗口計(jì)數(shù)的概念和關(guān)鍵代碼做了較為詳細(xì)解釋，如果還有不理解，請(qǐng)參考http://www.michael-noll.com/blog/2013/01/18/implementing-real-time-trending-topics-in-storm/的設(shè)計(jì)以及storm的源碼.
希望你了解了什么是實(shí)時(shí)計(jì)算 ：]
