【转】Twitter Storm: 在生产集群上运行topology

Twitter Storm: 在生产集群上运行topology

发表于 2011 年 10 月 07 日 由 xumingming

作者: xumingming | 可以转载, 但必须以超链接形式标明文章原始出处和作者信息及版权声明
网址: http://xumingming.sinaapp.com/185/twitter-storm-在生产集群上运行topology/

本文翻译自: https://github.com/nathanmarz/storm/wiki/Running-topologies-on-a-production-cluster 。

在生产集群上运行topology跟本地模式差不多。下面是步骤:

1)定义topology(如果是java的话, 用TopologyBuilder)

2) 使用StormSubmitter来把topology提交到集群。StormSubmitter的参数有:topology的名字,topology的配置对象,以及topology本身。比如:

帮助


1

2

3

4

5

Config conf = newConfig();

conf.setNumWorkers(20);

conf.setMaxSpoutPending(5000);

StormSubmitter.submitTopology("name",

                   conf, topology);

3) 创建一个包含你的程序代码以及你代码所依赖的依赖包的jar包(有关storm的jar包不用包括, 这些jar包会在工作节点上自动被添加到classpath里面去)。如果你使用maven, 那么插件:Maven Assembly Plugin可以帮你打包,只要把下面的配置加入你的pom.xml。

帮助


01

02

03

04

05

06

07

08

09

10

11

12

13

<plugin>

  <artifactId>maven-assembly-plugin</artifactId>

  <configuration>

    <descriptorRefs>

      <descriptorRef>jar-with-dependencies</descriptorRef>

    </descriptorRefs>

    <archive>

      <manifest>

        <mainClass>com.path.to.main.Class</mainClass>

      </manifest>

    </archive>

  </configuration>

</plugin>

然后运行mvn assembly:assembly就可以打包了. 再说一下,不用包括storm相关的jar包,它们会自动加到classpath里面。

4)用storm客户端去提交jar包:

帮助


1

storm jar allmycode.jar org.me.MyTopology arg1 arg2 arg3

storm jar 会把代码提交到集群并且配置StormSubmitter类以让它和正确的集群进行通信。在这个例子里面,上传jar包之后storm jar命令会调用org.me.MyTopology的main函数,参数是 arg1, arg2, arg3。关于如何配置你的storm客户端去和storm集群进行通信可以看下配置storm开发环境

常见配置

有很多topology级的配置可以设。这里有关于所有配置的清单, 以”TOPOLOGY”打头的配置是topology级别的配置,可以覆盖全局级别的配置。下面是一些比较常见的:

1)Config.TOPOLOGY_WORKERS:  这个设置用多少个工作进程来执行这个topology。比如,如果你把它设置成25, 那么集群里面一共会有25个java进程来执行这个topology的所有task。如果你的这个topology里面所有组件加起来一共有150的并行度,那么每个进程里面会有6个线程(150 / 25 = 6)。

2)Config.TOPOLOGY_ACKERS: 这个配置设置acker线程的数目。Ackers是Storm的可靠性API的一部分,关于storm的可靠性API可以看下:Twitter Storm如何保证消息不丢失

3)Config.TOPOLOGY_MAX_SPOUT_PENDING:  这个设置一个spout task上面最多有多少个没有处理的tuple(没有ack/failed)回复, 我们推荐你设置这个配置,以防止tuple队列爆掉。

4)Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS: 这个配置storm的tuple的超时时间  – 超过这个时间的tuple被认为处理失败了。这个设置的默认设置是30秒,对于大多数的topology都已经足够了。关于storm的可靠性API可以看看Twitter Storm如何保证消息不丢失

5)Config.TOPOLOGY_SERIALIZATIONS: 为了在你的tuple里面使用自定义类型,你可以用这个配置注册自定义serializer。

终止一个topology

要终止一个topology, 执行:

帮助


1

storm kill{stormname}

其中{stormname}是提交topology给storm集群的时候指定的名字。

storm不会马上终止topology。相反,它会先终止所有的spout,让它们不再发射任何新的tuple, storm会等Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS秒之后才杀掉所有的工作进程。这会给topology足够的时间来完成所有我们执行storm kill命令的时候还没完成的tuple。

更新一个运行中的topology

为了更新一个正在运行的topology, 唯一的选择是杀掉正在运行的topology然后重新提交一个新的。一个计划中的命令是实现一个storm swap命令来运行时更新topology, 并且保证前后两个topology不会同时在运行,同时保证替换所造成的“停机”时间最少。

监控topology

监控topology的最好的方法是使用Storm UI。Storm UI提供有关task里面发生的错误以及topology里面每个组件的吞吐量和性能方面的统计信息。同时你可以看看集群里面工作机器上面的日志。

时间: 2024-10-21 01:38:17

【转】Twitter Storm: 在生产集群上运行topology的相关文章

Storm在生产集群上运行Topology

一.步骤1.定义topology,若用java语言,使用TopologyBuilder来定义2.使用StormSubmitter来提交topology到集群中,所需参数需要topology名字,topology的参数配置,topology本身例:Config conf = new Config();conf.setNumWorkers(20);conf.setMaxSpoutPending(5000);StormSubmitter.submitTopology("mytopology"

storm学习之-在生产集群上运行topology

https://storm.apache.org/documentation/Setting-up-a-Storm-cluster.html -官方文档 http://xumingming.sinaapp.com/185/twitter-storm-在生产集群上运行topology/ --徐明明 http://blog.cheyo.net/84.html   --运行一个

用python + hadoop streaming 编写分布式程序(二) -- 在集群上运行与监控

写在前面 前文:用python + hadoop streaming 编写分布式程序(一) -- 原理介绍,样例程序与本地调试 为了方便,这篇文章里的例子均为伪分布式运行,一般来说只要集群配置得当,在伪分布式下能够运行的程序,在真实集群上也不会有什么问题. 为了更好地模拟集群环境,我们可以在mapred-site.xml中增设reducer和mapper的最大数目(默认为2,实际可用数目大约是CPU核数-1). 假设你为Hadoop安装路径添加的环境变量叫$HADOOP_HOME(如果是$HAD

将java开发的wordcount程序部署到spark集群上运行

1 package cn.spark.study.core; 2 3 import java.util.Arrays; 4 5 import org.apache.spark.SparkConf; 6 import org.apache.spark.api.java.JavaPairRDD; 7 import org.apache.spark.api.java.JavaRDD; 8 import org.apache.spark.api.java.JavaSparkContext; 9 impo

将java开发的wordcount程序提交到spark集群上运行

今天来分享下将java开发的wordcount程序提交到spark集群上运行的步骤. 第一个步骤之前,先上传文本文件,spark.txt,然用命令hadoop fs -put spark.txt /spark.txt,即可. 第一:看整个代码视图 打开WordCountCluster.java源文件,修改此处代码: 第二步: 打好jar包,步骤是右击项目文件----RunAs--Run Configurations 照图填写,然后开始拷贝工程下的jar包,如图,注意是拷贝那个依赖jar包,不是第

在集群上运行Spark应用的详细过程

在集群上运行Spark应用的详细过程: (1)用户通过spark-submit脚本提交应用 (2)spark-submit脚本启动驱动器程序,调用用户定义的main()方法 (3)驱动器程序与集群管理器通信,申请资源以启动执行器节点 (4)集群管理器为驱动器程序启动执行器节点 (5)驱动器进程执行用户应用中的操作.根据程序中所定义的对RDD的转换操作和行动操作,驱动器节点把动作以任务的形式发送到执行器进程 (6)任务在执行器程序中进行计算并保存结果 (7)如果驱动器程序的main()方法退出,或

在集群上运行caffe程序时如何避免Out of Memory

不少同学抱怨,在集群的GPU节点上运行caffe程序时,经常出现"Out of Memory"的情况.实际上,如果我们在提交caffe程序到某个GPU节点的同时,指定该节点某个比较空闲的gpu id,便可以避免"Out of Memory"的情况.步骤如下: 1. 在提交任务前,制作一个带有“nvidia-smi”命令的run_gpu.sh文件 #!/bin/bash #$ -V #$ -cwd #$ -j y #$ -S /bin/bash nvidia-smi

06、部署Spark程序到集群上运行

06.部署Spark程序到集群上运行 6.1 修改程序代码 修改文件加载路径 在spark集群上执行程序时,如果加载文件需要确保路径是所有节点能否访问到的路径,因此通常是hdfs路径地址.所以需要修改代码中文件加载路径为hdfs路径: ... //指定hdfs路径 sc.textFile("hdfs://mycluster/user/centos/1.txt") ... ? 修改master地址 SparkConf中需要指定master地址,如果是集群上运行,也可以不指定,运行时可以通

从认证到调度,K8s 集群上运行的小程序到底经历了什么?

作者 | 声东? 阿里云售后技术专家 导读:不知道大家有没有意识到一个现实:大部分时候,我们已经不像以前一样,通过命令行,或者可视窗口来使用一个系统了. 前言 现在我们上微博.或者网购,操作的其实不是眼前这台设备,而是一个又一个集群.通常,这样的集群拥有成百上千个节点,每个节点是一台物理机或虚拟机.集群一般远离用户,坐落在数据中心.为了让这些节点互相协作,对外提供一致且高效的服务,集群需要操作系统.Kubernetes 就是这样的操作系统. 比较 Kubernetes 和单机操作系统,Kuber