欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

实时计算知多少?

程序员文章站 2024-03-05 08:11:42
实时计算是什么? 请看下面的图: 我们以热卖产品的统计为例,看下传统的计算手段: 1将用户行为、log等信息清洗后保存在数据库中. 2将订单信息保存在数据库中....

实时计算是什么?
请看下面的图:

实时计算知多少?

我们以热卖产品的统计为例,看下传统的计算手段:

1将用户行为、log等信息清洗后保存在数据库中.
2将订单信息保存在数据库中.
3利用触发器或者协程等方式建立本地索引,或者远程的独立索引.
4join订单信息、订单明细、用户信息、商品信息等等表,聚合统计20分钟内热卖产品,并返回top-10.
5web或app展示.

这是一个假想的场景,但假设你具有处理类似场景的经验,应该会体会到这样一些问题和难处:

1、水平扩展问题(scale-out)
显然,如果是一个具有一定规模的电子商务网站,数据量都是很大的。而交易信息因为涉及事务,所以很难直接舍弃关系型数据库的事务能力,迁移到具有更好的scale-out能力的nosql数据库中。

那么,一般都会做sharding。历史数据还好说,我们可以按日期来归档,并可以通过批处理式的离线计算,将结果缓存起来。
但是,这里的要求是20分钟内,这很难。

2、性能问题
这个问题,和scale-out是一致的,假设我们做了sharding,因为表分散在各个节点中,所以我们需要多次入库,并在业务层做聚合计算。

问题是,20分钟的时间要求,我们需要入库多少次呢?
10分钟呢?
5分钟呢?
实时呢?
而且,业务层也同样面临着单点计算能力的局限,需要水平扩展,那么还需要考虑一致性的问题。
所以,到这里一切都显得很复杂。

3、业务扩展问题
假设我们不仅仅要处理热卖商品的统计,还要统计广告点击、或者迅速根据用户的访问行为判断用户特征以调整其所见的信息,更加符合用户的潜在需求等,那么业务层将会更加复杂。

也许你有更好的办法,但实际上,我们需要的是一种新的认知:

这个世界发生的事,是实时的。
所以我们需要一种实时计算的模型,而不是批处理模型。
我们需要的这种模型,必须能够处理很大的数据,所以要有很好的scale-out能力,最好是,我们都不需要考虑太多一致性、复制的问题。

那么,这种计算模型就是实时计算模型,也可以认为是流式计算模型。

现在假设我们有了这样的模型,我们就可以愉快地设计新的业务场景:

转发最多的微博是什么?
最热卖的商品有哪些?
大家都在搜索的热点是什么?
我们哪个广告,在哪个位置,被点击最多?

或者说,我们可以问:

这个世界,在发生什么?

最热的微博话题是什么?

我们以一个简单的滑动窗口计数的问题,来揭开所谓实时计算的神秘面纱。

假设,我们的业务要求是:

统计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<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);
 }

上面的代码可能有点抽象,看下这个图就明白了,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仅仅是为了防盗版 : ].

实时计算知多少?

以上就是本文的全部内容,希望大家喜欢。