Storm-源码分析-Stats (backtype.storm.stats)-低调大师

Storm-源码分析-Stats (backtype.storm.stats)

2017-05-01 577

会发现, 现在storm里面有两套metrics系统, metrics framework和stats framework

并且在所有地方都是同时注册两套, 貌似准备用metrics来替代stats, 但当前版本UI仍然使用stats

这个模块统计的数据怎么被使用,

1. 在worker中, 会定期调用do-executor-heartbeats去往zk同步hb
可以看到, stats也会作为hb的一部分被同步到zk上

(defnk do-executor-heartbeats [worker :executors nil]
  ;; stats is how we know what executors are assigned to this worker 
  (let [stats (if-not executors
                  (into {} (map (fn [e] {e nil}) (:executors worker)))
                  (->> executors
                    (map (fn [e] {(executor/get-executor-id e) (executor/render-stats e)}))
                    (apply merge)))
        zk-hb {:storm-id (:storm-id worker)
               :

executor-stats stats

               :uptime ((:uptime worker))
               :time-secs (current-time-secs)
               }]
    ;; do the zookeeper heartbeat
    (.worker-heartbeat! (:storm-cluster-state worker) (:storm-id worker) (:assignment-id worker) (:port worker) zk-hb)    
    ))

2. 现在任何人都可以通过nimbus的thrift接口来得到相关信息

(^TopologyInfo getTopologyInfo [this ^String storm-id]
   beats (.executor-beats storm-cluster-state storm-id (:executor->node+port assignment))
   stats (:stats heartbeat))

3. 最直接的用户就是storm UI, 在准备topology page的时候, 就会调用getTopologyInfo来获取数据

(defn topology-page [id window include-sys?]
  (with-nimbus nimbus
    (let [summ (.getTopologyInfo ^Nimbus$Client nimbus id)]
)

Stats

这个模块用于spout和bolt来抽样统计数据, 需要统计的具体metics如下

(def COMMON-FIELDS [:emitted :transferred])
(defrecord CommonStats [emitted transferred rate])

(def BOLT-FIELDS [:acked :failed :process-latencies :executed :execute-latencies])
;;acked and failed count individual tuples
(defrecord BoltExecutorStats [common acked failed process-latencies executed execute-latencies])

(def SPOUT-FIELDS [:acked :failed :complete-latencies])
;;acked and failed count tuple completion
(defrecord SpoutExecutorStats [common acked failed complete-latencies])

抽样的比例在storm-conf, TOPOLOGY_STATS_SAMPLE_RATE, 配置

为什么统计时每次加rate, 而不是加1?

因为这里的统计是抽样的, 所以如果抽样比例是10%, 那么发现一个, 应该加1/(10%), 10个
(defn sampling-rate [conf]
  (->> (conf TOPOLOGY-STATS-SAMPLE-RATE)
       (/ 1)
       int))

然后统计是基于时间窗口的, 底下是对应默认的bucket和时间窗口的定义

(def NUM-STAT-BUCKETS 20) ;;bucket数
;; 10 minutes, 3 hours, 1 day ;;定义3种时间窗口
(def STAT-BUCKETS [30 540 4320]) ;;bucket大小分别是30，540，4320秒

核心数据结构是RollingWindowSet, 包含:
统计数据需要的函数, updater extractor, 之所以治理也需要是因为需要统计all-time
一组rolling windows, 默认是3个时间窗, 10 minutes, 3 hours, 1 day
all-time, 在完整的时间区间上的统计结果

(defrecord RollingWindowSet [updater extractor windows all-time])
(defn rolling-window-set [updater merger extractor num-buckets & bucket-sizes]
  (RollingWindowSet. updater extractor (dofor [s bucket-sizes] (rolling-window updater merger extractor s num-buckets)) nil)
  )

继续看看rolling window的定义,
核心数据, buckets, hashmap, {streamid, data}, 初始化为{}
统计data需要的函数, updater merger extractor
时间窗口, buckets大小和buckets个数

(defrecord RollingWindow [updater merger extractor bucket-size-secs num-buckets buckets])
(defn rolling-window [updater merger extractor bucket-size-secs num-buckets]
  (RollingWindow. updater merger extractor bucket-size-secs num-buckets {}))

1. mk-stats

在mk-executedata的时候需要创建stats

mk-executor-stats <> (sampling-rate storm-conf)

;; TODO: refactor this to be part of an executor-specific map
(defmethod mk-executor-stats :spout [_ rate]
  (stats/mk-spout-stats rate))
(defmethod mk-executor-stats :bolt [_ rate]
  (stats/mk-bolt-stats rate))

第一个参数忽略, 其实就是分别调用stats/mk-spout-stats或stats/mk-bolt-stats, 可见就是对于每个需要统计的数据, 创建一个rolling-windows-set

(defn- mk-common-stats [rate]
  (CommonStats. (atom (apply keyed-counter-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                (atom (apply keyed-counter-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                rate
                ))

(defn mk-bolt-stats [rate]
  (BoltExecutorStats. (mk-common-stats rate)
                  (atom (apply keyed-counter-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                  (atom (apply keyed-counter-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                  (atom (apply keyed-avg-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                  (atom (apply keyed-counter-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                  (atom (apply keyed-avg-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                  ))

(defn mk-spout-stats [rate]
  (SpoutExecutorStats. (mk-common-stats rate)
                   (atom (apply keyed-counter-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                   (atom (apply keyed-counter-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                   (atom (apply keyed-avg-rolling-window-set NUM-STAT-BUCKETS STAT-BUCKETS))
                   ))

2. 数据更新

(defn spout-acked-tuple! [^SpoutExecutorStats stats stream latency-ms]
  (update-executor-stat! stats :acked stream (stats-rate stats))
  (update-executor-stat! stats :complete-latencies stream latency-ms)
  )

(defmacro update-executor-stat! [stats path & args]
  (let [path (collectify path)]
    `(swap! (-> ~stats ~@path) update-rolling-window-set ~@args)
    ))

就以update-executor-stat! stats :acked stream (stats-rate stats)为例子看看怎么做的?

SpoutExecutorStats取出用于记录spout acked情况的rolling-windows-set
然后使用update-rolling-window-set来swap这个atom

来看看记录acked的rolling-windows-set是如何定义的?

keyed-counter-rolling-window-set, 预定义了updater merger extractor
updater, incr-val [amap key amt], 把给定的值amt加到amap的对应的key的value上
merger, (partial merge-with +), 用+作为map merge的逻辑, 即出现相同key则相加
extractor, counter-extract, (if v v {}), 有则返回, 无则返回{}
windows, rolling-window的list
all-time, 初始化为nil

(defn keyed-counter-rolling-window-set [num-buckets & bucket-sizes]
  (apply rolling-window-set incr-val (partial merge-with +) counter-extract num-buckets bucket-sizes))

好, 下面就看看, 当spout-acked-tuple!时更新:acked时, 如何update的?

首先更新每个rolling-window, 并把更新过的rolling-window-set更新到:windows
并且更新:all-time, (apply (:updater rws) (:all-time rws) args)
updated, incr-val [amap key amt]
args, steamid, rate
all-time, 是用来记录整个时间区间上的, 某个stream的统计情况

(defn update-rolling-window-set
  ([^RollingWindowSet rws & args]
     (let [now (current-time-secs)
           new-windows (dofor [w (:windows rws)]
                         (apply update-rolling-window w now args))]
       (assoc rws :windows new-windows :all-time (apply (:updater rws) (:all-time rws) args))
       )))

看下如何更新某个rolling-windw
根据now算出当前属于哪个bucket, time-bucket
取出buckets, 并使用:updater更新相应的bucket, 这里的操作仍然是把rate叠加到streamid的value上

(defn update-rolling-window
  ([^RollingWindow rw time-secs & args]
     ;; this is 2.5x faster than using update-in...
     (let [time-bucket (curr-time-bucket time-secs (:bucket-size-secs rw))
           buckets (:buckets rw)
           curr (get buckets time-bucket)           
           curr (apply (:updater rw) curr args)
           ]
       (assoc rw :buckets (assoc buckets time-bucket curr))
       )))

本文章摘自博客园，原文发布日期：2013-07-29

微信关注我们

原文链接：https://yq.aliyun.com/articles/85668

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

《Flume日志收集与MapReduce模式》一1.5　Flume事件

本节书摘来自华章出版社《Flume日志收集与MapReduce模式》一书中的第1章，第1.5节，作者［美］史蒂夫·霍夫曼（Steve Hoffman）斯里纳特·佩雷拉（Srinath Perera），更多章节内容可以访问云栖社区“华章计算机”公众号查看 1.5　Flume事件 Flume传输的基本的数据负载叫作事件。事件由0个或多个头与体组成。头是一些键值对，可用于路由判定或是承载其他的结构化信息（比如说事件的时间戳或是发出事件的服务器主机名）。你可以将其看作是与HTTP头完成相同的功能——传递与体不同的额外信息的方式。体是个字节数组，包含了实际的负载。如果输入由日志文件组成，那么该数组就非常类似于包含了单行文本的UTF-8编码的字符串。 Flume可能会自动添加头（比如，源添加了数据来自的主机名或是创建了事件时间戳），不过体基本上

2017-05-02

470

1.2　在机器学习中应用Spark计算基于RDD和内存处理的创新功能，Apache Spark真正使得分布式计算对于数据科学家和机器学习专业人员来说简便易用。Apache Spark团队表示：Apache Spark基于Mesos 集群管理器运行，使其可以与Hadoop以及其他应用共享资源。因此，Apache Spark可以从任何Hadoop输入源（如HDFS）中读取数据。 Apache Spark计算模型非常适合机器学习中的分布式计算。特别是在快速交互式机器学习、并行计算和大型复杂模型情境下，Apache Spark无疑可以发挥其卓越效能。 Spark开发团队表示，Spark的哲学是使数据科学家和机器学习专业人员的生活更加轻松和高效。因此，Apache Spark拥有以下特点：拥有详细说明文档，表达清晰的API 强大的专业领域库易于

2017-05-02

609

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。