Storm源码浅析之topology的提交-低调大师

Storm源码浅析之topology的提交

2017-05-01 609

最近一直在读twitter开源的这个分布式流计算框架——storm的源码，还是有必要记录下一些比较有意思的地方。我按照storm的主要概念进行组织，并且只分析我关注的东西，因此称之为浅析。

一、介绍
Storm的开发语言主要是Java和Clojure，其中Java定义骨架，而Clojure编写核心逻辑。源码统计结果：

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>     180 text files.
     177 unique files.
       7 files ignored.

http://cloc.sourceforge.net v 1.55  T=1.0 s (171.0 files/s, 46869.0 lines/s)
——————————————————————————-
Language                     files          blank        comment           code
——————————————————————————-
Java                           125           5010           2414          25661
Lisp                            33            732            283           4871
Python                           7            742            433           4675
CSS                              1             12             45           1837
ruby                             2             22              0            104
Bourne Shell                     1              0              0              6
Javascript                       2              1             15              6
——————————————————————————-
SUM:                           171           6519           3190          37160
——————————————————————————-

Java代码25000多行，而Clojure(Lisp)只有4871行，说语言不重要再次证明是扯淡。

二、Topology和Nimbus
    Topology是storm的核心理念，将spout和bolt组织成一个topology，运行在storm集群里，完成实时分析和计算的任务。这里我主要想介绍下topology部署到storm集群的大概过程。提交一个topology任务到Storm集群是通过StormSubmitter.submitTopology方法提交：

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>StormSubmitter.submitTopology(name, conf, builder.createTopology());

我们将topology打成jar包后，利用bin/storm这个python脚本，执行如下命令：

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>bin/storm jar xxxx.jar com.taobao.MyTopology args

将jar包提交给storm集群。storm脚本会启动JVM执行Topology的main方法，执行submitTopology的过程。而submitTopology会将jar文件上传到nimbus，上传是通过socket传输。在storm这个python脚本的jar方法里可以看到：

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>def jar(jarfile, klass, args):
   exec_storm_class(
        klass,
        jvmtype=“-client“,
        extrajars=[jarfile, CONF_DIR, STORM_DIR + “/bin“],
        args=args,
        prefix=“export STORM_JAR=“ + jarfile + “;“)

将jar文件的地址设置为环境变量STORM_JAR，这个环境变量在执行submitTopology的时候用到：

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>//StormSubmitter.java
private static void submitJar(Map conf) {
if(submittedJar==null) {
LOG.info(“Jar not uploaded to master yet. Submitting jar

“);
            String localJar = System.getenv(“STORM_JAR“);
            submittedJar = submitJar(conf, localJar);
        } else {
            LOG.info(“Jar already uploaded to master. Not submitting jar.“);
        }
    }

通过环境变量找到jar包的地址，然后上传。利用环境变量传参是个小技巧。

其次，nimbus在接收到jar文件后，存放到数据目录的inbox目录，nimbus数据目录的结构：

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>-nimbus
     -inbox
         -stormjar-57f1d694-2865-4b3b-8a7c-99104fc0aea3.jar
         -stormjar-76b4e316-b430-4215-9e26-4f33ba4ee520.jar

     -stormdist
        -storm-id
           -stormjar.jar
           -stormconf.ser
           -stormcode.ser

其中inbox用于存放提交的jar文件，每个jar文件都重命名为stormjar加上一个32位的UUID。而stormdist存放的是启动topology后生成的文件，每个topology都分配一个唯一的id，ID的规则是“name-计数-时间戳”。启动后的topology的jar文件名命名为storm.jar ,而它的配置经过java序列化后存放在stormconf.ser文件，而stormcode.ser是将topology本身序列化后存放的文件。这些文件在部署的时候，supervisor会从这个目录下载这些文件，然后在supervisor本地执行这些代码。
进入重点，topology任务的分配过程(zookeeper路径说明忽略root):
1.在zookeeper上创建/taskheartbeats/{storm id} 路径，用于任务的心跳检测。storm对zookeeper的一个重要应用就是利用zk的临时节点做存活检测。task将定时刷新节点的时间戳，然后nimbus会检测这个时间戳是否超过timeout设置。
2.从topology中获取bolts,spouts设置的并行数目以及全局配置的最大并行数，然后产生task id列表，如[1 2 3 4]
3.在zookeeper上创建/tasks/{strom id}/{task id}路径，并存储task信息
4.开始分配任务（内部称为assignment)，具体步骤：
(1)从zk上获得已有的assignment(新的toplogy当然没有了）
(2)查找所有可用的slot，所谓slot就是可用的worker，在所有supervisor上配置的多个worker的端口。
(3)将任务均匀地分配给可用的worker，这里有两种情况：
(a)task数目比worker多，例如task是[1 2 3 4],可用的slot只有[host1:port1 host2:port1]，那么最终是这样分配

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>{1: [host1:port1] 2 : [host2:port1]
3 : [host1:port1] 4 : [host2:port1]}

，可以看到任务平均地分配在两个worker上。
(b)如果task数目比worker少，例如task是[1 2]，而worker有[host1:port1 host1:port2 host2:port1 host2:port2]，那么首先会将woker排序，*将不同host间隔排列，保证task不会全部分配到同一个worker上，也就是将worker排列成

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>[host1:port1 host2:port1 host1:port2 host2:port2]

，然后分配任务为

<!–

Code highlighting produced by Actipro CodeHighlighter (freeware)
http://www.CodeHighlighter.com/

–>{1: host1:port1 , 2 : host2:port2}

(4)记录启动时间
(5)判断现有的assignment是否跟重新分配的assignment相同，如果相同，不需要变更，否则更新assignment到zookeeper的/assignments/{storm id}上。
5.启动topology，所谓启动，只是将zookeeper上/storms/{storm id}对应的数据里的active设置为true。

6.nimbus会检查task的心跳，如果发现task心跳超过超时时间，那么会重新跳到第4步做re-assignment。

本文来源于"阿里中间件团队播客",原文发表时间" 2011-12-02 "

微信关注我们

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

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

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

《深入理解ElasticSearch》——3.5　深入理解数据处理

本节书摘来自华章计算机《深入理解ElasticSearch》一书中的第3章，第3.5节,作者：[美] 拉斐尔·酷奇（Rafa Ku）　马雷克·罗戈任斯基（Marek Rogoziński）更多章节内容可以访问云栖社区“华章计算机”公众号查看。 3.5　深入理解数据处理刚开始使用ElasticSearch的时候，各种搜索方式和查询类型往往会令人感到头疼。查询类型之间行为各异，尽管有些差别非常细微，例如范围查询和前缀查询。了解这些差别对理解查询的工作原理至关重要，尤其是要在ElasticSearch提供的默认查询类型之外做些额外事情的时候，例如，处理多语言信息。3.5.1　输入并不总是进行文本分析在讨论查询分析之前，我们先使用以下命令创建一个索引：如你所见，这个索引非常简单。文档只包含一个字段，该字段使用snowball分析器进行处理。

2017-05-02

650

本节书摘来异步社区《HBase权威指南》一书中的第3章，第3.6节，作者：【美】Lars George 译者：代志远 , 刘佳 , 蒋杰责编：杨海玲，更多章节内容可以访问云栖社区“异步社区”公众号查看。 3.6　各种特性在深入介绍客户端可以利用的特性之前，让我们先介绍一下HBase和其客户端API提供的各种特性或功能。 3.6.1　HTable的实用方法客户端API是由HTable类的实例提供的，用户可以用它来操作HBase表。除了之前提到过的一些主要特性外，还有以下一些值得注意的方法。 void close() 这个方法之前提到过，不过为了它的完整性和重要性，我们在这儿要重新讨论一下。用户使用完一个HTable实例之后，需要调用一次close()。这个方法会刷写所有客户端缓冲的写操作：close()方法会隐式调用flushCache()方法。 byte[] getTableName() 这是一个获取表名称的快捷方法。 Configuration getConfiguration() 这个方法允许用户访问HTable实例中使用的配置。因为得到的是Configuration实...

2017-05-02

607

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

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

Spring

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

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。