歡迎光臨
每天分享高質量文章

你瞭解實時計算嗎?

作者:foreach_break

網址:http://www.cnblogs.com/foreach-break/p/what-is-real-time-computing-and-how.html

點擊“閱讀原文”可查看本文網頁版

實時計算是什麼?

請看下麵的圖:

我們以熱賣產品的統計為例,看下傳統的計算手段:

  1. 將用戶行為、log等信息清洗後儲存在資料庫中.
  2. 將訂單信息儲存在資料庫中.
  3. 利用觸發器或者協程等方式建立本地索引,或者遠程的獨立索引.
  4. join訂單信息、訂單明細、用戶信息、商品信息等等表,聚合統計20分鐘內熱賣產品,並傳回top-10.
  5. web或app展示.


這是一個假想的場景,但假設你具有處理類似場景的經驗,應該會體會到這樣一些問題和難處:

  1. 水平擴展問題(scale-out)
    顯然,如果是一個具有一定規模的電子商務網站,資料量都是很大的。而交易信息因為涉及事務,所以很難直接捨棄關係型資料庫的事務能力,遷移到具有更好的scale-out能力的NoSQL資料庫中。

    那麼,一般都會做sharding。歷史資料還好說,我們可以按日期來歸檔,並可以通過批處理式的離線計算,將結果快取起來。
    但是,這裡的要求是20分鐘內,這很難。
  2. 性能問題
    這個問題,和scale-out是一致的,假設我們做了sharding,因為表分散在各個節點中,所以我們需要多次入庫,併在業務層做聚合計算。

    問題是,20分鐘的時間要求,我們需要入庫多少次呢?
    10分鐘呢?
    5分鐘呢?
    實時呢?


    而且,業務層也同樣面臨著單點計算能力的局限,需要水平擴展,那麼還需要考慮一致性的問題。


    所以,到這裡一切都顯得很複雜。

  3. 業務擴展問題
    假設我們不僅僅要處理熱賣商品的統計,還要統計廣告點擊、或者迅速根據用戶的訪問行為判斷用戶特征以調整其所見的信息,更加符合用戶的潛在需求等,那麼業務層將會更加複雜。

也許你有更好的辦法,但實際上,我們需要的是一種新的認知:

這個世界發生的事,是實時的。


所以我們需要一種實時計算的模型,而不是批處理模型。


我們需要的這種模型,必須能夠處理很大的資料,所以要有很好的scale-out能力,最好是,我們都不需要考慮太多一致性、複製的問題。

那麼,這種計算模型就是實時計算模型,也可以認為是流式計算模型。

現在假設我們有了這樣的模型,我們就可以愉快地設計新的業務場景:

  1. 轉發最多的微博是什麼?
  2. 最熱賣的商品有哪些?
  3. 大家都在搜索的熱點是什麼?
  4. 我們哪個廣告,在哪個位置,被點擊最多?

或者說,我們可以問:

這個世界,在發生什麼?

最熱的微博話題是什麼?

我們以一個簡單的滑動視窗計數的問題,來揭開所謂實時計算的神秘面紗。

假設,我們的業務要求是:

統計20分鐘內最熱的10個微博話題。

解決這個問題,我們需要考慮:

  1. 資料源
    這裡,假設我們的資料,來自微博長連接推送的話題。
  2. 問題建模
    我們認為的話題是#號擴起來的話題,最熱的話題是此話題出現的次數比其它話題都要多。
    比如:@foreach_break : 你好,#世界#,我愛你,#微博#。


    “世界”和“微博”就是話題。

  3. 計算引擎
    我們採用storm。
  4. 定義時間

如何定義時間?

時間的定義是一件很難的事情,取決於所需的精度是多少。


根據實際,我們一般採用tick來表示時刻這一概念。

在storm的基礎設施中,executor啟動階段,採用了定時器來觸發“過了一段時間”這個事件。


如下所示:

(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)]]
              )))))))

之前的博文中,已經詳細分析了這些基礎設施的關係,不理解的童鞋可以翻看前面的文章。

每隔一段時間,就會觸發這樣一個事件,當流的下游的bolt收到一個這樣的事件時,就可以選擇是增量計數還是將結果聚合併發送到流中。

bolt如何判斷收到的tuple表示的是“tick”呢?


負責管理bolt的executor執行緒,從其訂閱的訊息佇列消費訊息時,會呼叫到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());
}

結合上面的setup-tick!的clojure代碼,我們可以知道SYSTEM_TICK_STREAM_ID在定時事件的回呼中就以建構式的引數傳遞給了tuple,那麼SYSTEM_COMPONENT_ID是如何來的呢?


可以看到,下麵的代碼中,SYSTEM_TASK_ID同樣傳給了tuple:

;; 請註意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);
        }
    }

滑動視窗

有了上面的基礎設施,我們還需要一些手段來完成“工程化”,將設想變為現實。

這裡,我們看看Michael G. Noll的滑動視窗設計。

Topology


    String spoutId = "wordGenerator";
    String counterId = "counter";
    String intermediateRankerId = "intermediateRanker";
    String totalRankerId = "finalRanker";    // 這裡,假設TestWordSpout就是我們發送話題tuple的源
    builder.setSpout(spoutId, new TestWordSpout(), 5);    // RollingCountBolt的時間視窗為9秒鐘,每3秒發送一次統計結果到下游
    builder.setBolt(counterId, new RollingCountBolt(9, 3), 4).fieldsGrouping(spoutId, new Fields("word"));    // IntermediateRankingsBolt,將完成部分聚合,統計出top-n的話題
    builder.setBolt(intermediateRankerId, new IntermediateRankingsBolt(TOP_N), 4).fieldsGrouping(counterId, new Fields(        "obj"));        // TotalRankingsBolt, 將完成完整聚合,統計出top-n的話題
    builder.setBolt(totalRankerId, new TotalRankingsBolt(TOP_N)).globalGrouping(intermediateRankerId);

上面的topology設計如下:


將聚合計算與時間結合起來

前文,我們敘述了tick事件,回呼中會觸發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來了,將時間視窗內的統計結果發送,並讓視窗滾動
      emitCurrentWindowCounts();
    }    else {      // 常規tuple,對話題計數即可
      countObjAndAck(tuple);
    }
  }  // obj即為話題,增加一個計數 count++
  // 註意,這裡的速度基本取決於流的速度,可能每秒百萬,也可能每秒幾十.
  // 記憶體不足? bolt可以scale-out.
  private void countObjAndAck(Tuple tuple) {
    Object obj = tuple.getValue(0);
    counter.incrementCount(obj);
    collector.ack(tuple);
  }  
  // 將統計結果發送到下游
	private void emitCurrentWindowCounts() {
	  Map 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);
	}

上面的代碼可能有點抽象,看下這個圖就明白了,tick一到,視窗就滾動:


IntermediateRankingsBolt & TotalRankingsBolt:

  public final void execute(Tuple tuple, BasicOutputCollector collector) {    if (TupleUtils.isTick(tuple)) {
      getLogger().debug("Received tick tuple, triggering emit of current rankings");      // 將聚合併排序的結果發送到下游
      emitRankings(collector);
    }    else {      // 聚合併排序
      updateRankingsWithTuple(tuple);
    }
  }

其中,IntermediateRankingsBolt和TotalRankingsBolt的聚合排序方法略有不同:

IntermediateRankingsBolt的聚合排序方法:

// IntermediateRankingsBolt的聚合排序方法:
  @Override
  void updateRankingsWithTuple(Tuple tuple) {    // 這一步,將話題、話題出現的次數提取出來
    Rankable rankable = RankableObjectWithFields.from(tuple);    // 這一步,將話題出現的次數進行聚合,然後重排序所有話題
    super.getRankings().updateWith(rankable);
  }

TotalRankingsBolt的聚合排序方法:

// TotalRankingsBolt的聚合排序方法
  @Override
  void updateRankingsWithTuple(Tuple tuple) {  // 提出來自IntermediateRankingsBolt的中間結果
    Rankings rankingsToBeMerged = (Rankings) tuple.getValue(0);  // 聚合併排序
    super.getRankings().updateWith(rankingsToBeMerged);  // 去0,節約記憶體
    super.getRankings().pruneZeroCounts();
  }

而重排序方法比較簡單粗暴,因為只求前N個,N不會很大:

  private void rerank() {
    Collections.sort(rankedItems);
    Collections.reverse(rankedItems);
  }

結語

下圖可能就是我們想要的結果,我們完成了t0 – t1時刻之間的熱點話題統計,其中的foreach_break僅僅是為了防盜版 : ].

文中對滑動視窗計數的概念和關鍵代碼做了較為詳細解釋,如果還有不理解,請參考storm的原始碼.

希望你瞭解了什麼是實時計算 :]

赞(0)

分享創造快樂