storm-0.8.2集群模式安装部署

2024-06-04 09:18

本文主要是介绍storm-0.8.2集群模式安装部署,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

标签(空格分隔): storm


hadoop,spark,kafka交流群:224209501

集群:
一台主机用来运行nimbus,另外两台运行superior。

安装及测试:

1,安装ZeroMQ
2,安装jzmq
3,安装Python
4,安装storm
5,配置storm
6,启动storm
7,测试storm

1,安装依赖

sudo yum install uuid*
sudo yum install libtool
sudo yum install libuuid 
sudo yum install libuuid-devel
sudo yum install libtool*

2,安装zeromq

 tar -zxvf zeromq-2.1.7.tar.gz -C /opt/storm/./configuremakesudo make install

3,安装jzmq

unzip jzmq-master.zip -d /opt/storm/
./autogen.sh 
./configure
make
make install

4,配置storm

1,说明一下:storm.local.dir表示storm要用到的本地目录。nimbus.host表示哪一台机器是master,即nimbus。storm.zookeeper.port表示zookeeper的端口号,这里一定要与zookeeper配置端口号一致,否则会出现通信错误。当然你也可以配置superevisor.slot.ports,supervisor.slots.ports表示supervisor节点的槽数,就是最多能跑几个worker进程(每个spout或者bolt默认只启动一个worker,但是可以通过conf修改成多个)
2,java.library.path这是storm所依赖的本地依赖(ZeroMQ和JZMQ)的加载地址,默认的是:/usr/local/lib:/opt/local/lib:/usr/lib,大多数情况下是对的,所以你应该不用更改这个配置。
注意事项:
1,配置时一定注意在每一项的开始时是要加空格(最好加两个空格),冒号之后也要加空格,否则storm不认识这个配置文件。
2,在目录/usr/tmp下面增加storm文件夹。
在storm.yaml文件中配置如下内容:

 storm.zookeeper.servers:- "miaodonghua.host"
#     - "server2"
# nimbus.host: "miaodonghua.host"

5,启动nimbus

bin/storm nimbus

nimbun启动成功.png-9.3kB

6,启动storm的web UI

经测试,StormUI必须和Storm Nimbus部署在同一台机器上,否则UI将无法正常工作,因为UI进程会检查本机是否存在Nimbus链接。

nohup bin/storm ui & 

storm ui.png-11kB
浏览器中输入下面链接及端口号

http://miaodonghua.host:8080///浏览器

主节点ui.png-60.9kB

7,启动从节点

在两台supervisor机器中启动

nohup bin/storm supervisor &

supervisor启动成功.png-9.6kB
stom ui supervisor.png-49.3kB

8,运行

bin/storm jar /opt/storm/storm-0.8.2/lifeCycle.jar  cn.storm.topology.TopoMain
bin/storm list

stormlist.png-21.5kB

http://miaodonghua.host:8080

执行成功.png-45.6kB
lifecycle.png-55.7kB

9,storm-start-master

下载storm-start:
storm-starter-0.7.0.zip
进入主目录修改m2-pom.xml(将twitter4j-core和twitter4j-stream替换为下面的部分)

<dependency><groupId>org.twitter4j</groupId><artifactId>twitter4j-core</artifactId>
<version>[2.2,)</version>
</dependency>
<dependency><groupId>org.twitter4j</groupId><artifactId>twitter4j-stream</artifactId><version>[2.2,)</version>
</dependency>

编译

mvn -f m2-pom.xml package

编译成功.png-23.3kB
复制 storm-starter目录下的m2_pom.xml 为pom.xml,放在与m2_pom.xml同一目录下
打jar包

mvn jar:jar

打成jar包.png-7.3kB
如果需要对工程代码进行修改,可以导入eclipse

mvn eclipse:eclipse

生成eclipse成功.png-10.1kB

10,源码内容如下

1,编写RandomWordSpout

package cn.storm.spout;import java.util.Map;
import java.util.Random;import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
/*** 随机从String数组当中读取一个单词发送给下一个bolt* @author Administrator**/
public class RandomWordSpout extends BaseRichSpout {private static final long serialVersionUID = -4287209449750623371L;private static final Log log = LogFactory.getLog(RandomWordSpout.class);private SpoutOutputCollector collector;private String[] words = new String[]{"storm", "hadoop", "hive", "flume"};private Random random = new Random();public RandomWordSpout() {log.warn("&&&&&&&&&&&&&&&&& RandomWordSpout constructor method invoked");}@Overridepublic void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {log.warn("################# RandomWordSpout open() method invoked");this.collector = collector;}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {log.warn("################# RandomWordSpout declareOutputFields() method invoked");declarer.declare(new Fields("str"));}@Overridepublic void nextTuple() {log.warn("################# RandomWordSpout nextTuple() method invoked");Utils.sleep(500);String str = words[random.nextInt(words.length)];collector.emit(new Values(str));}@Overridepublic void activate() {log.warn("################# RandomWordSpout activate() method invoked");}@Overridepublic void deactivate() {log.warn("################# RandomWordSpout deactivate() method invoked");}
}

2,编写TransferBolt

package cn.storm.bolt;import java.util.Map;import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;/*** 将数据做简单的 传递的Bolt* @author Administrator**/
public class TransferBolt extends BaseBasicBolt {private static final long serialVersionUID = 4223708336037089125L;private static final Log log = LogFactory.getLog(TransferBolt.class);public TransferBolt() {log.warn("&&&&&&&&&&&&&&&&& TransferBolt constructor method invoked");}@Overridepublic void prepare(Map stormConf, TopologyContext context) {log.warn("################# TransferBolt prepare() method invoked");}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {log.warn("################# TransferBolt declareOutputFields() method invoked");declarer.declare(new Fields("word"));}@Overridepublic void execute(Tuple input, BasicOutputCollector collector) {log.warn("################# TransferBolt execute() method invoked");String word = input.getStringByField("str");collector.emit(new Values(word));}}

3,编写WriterBolt

package cn.storm.bolt;import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;
/*** 将接收到的单词写入到一个文件当中* @author Administrator**/
public class WriterBolt extends BaseBasicBolt {private static final long serialVersionUID = -6586283337287975719L;private static final Log log = LogFactory.getLog(WriterBolt.class);private FileWriter writer = null;public WriterBolt() {log.warn("&&&&&&&&&&&&&&&&& WriterBolt constructor method invoked");}@Overridepublic void prepare(Map stormConf, TopologyContext context) {log.warn("################# WriterBolt prepare() method invoked");try {writer = new FileWriter("/home/" + this);} catch (IOException e) {log.error(e);throw new RuntimeException(e);}}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {log.warn("################# WriterBolt declareOutputFields() method invoked");}@Overridepublic void execute(Tuple input, BasicOutputCollector collector) {log.warn("################# WriterBolt execute() method invoked");String s = input.getString(0);try {writer.write(s);writer.write("\n");writer.flush();} catch (IOException e) {log.error(e);throw new RuntimeException(e);}}}

4,编写TopoMain

package cn.storm.topology;import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import cn.itcast.storm.bolt.TransferBolt;
import cn.itcast.storm.bolt.WriterBolt;
import cn.itcast.storm.spout.RandomWordSpout;public class TopoMain {private static final Log log = LogFactory.getLog(TopoMain.class);/*** @param args*/public static void main(String[] args) throws Exception {TopologyBuilder builder = new TopologyBuilder();builder.setSpout("random", new RandomWordSpout(), 2);builder.setBolt("transfer", new TransferBolt(), 4).shuffleGrouping("random");builder.setBolt("writer", new WriterBolt(), 4).fieldsGrouping("transfer", new Fields("word"));Config conf = new Config();conf.setNumWorkers(2);conf.setDebug(true);log.warn("$$$$$$$$$$$ submitting topology...");StormSubmitter.submitTopology("life-cycle", conf, builder.createTopology());log.warn("$$$$$$$4$$$ topology submitted !");}}

这篇关于storm-0.8.2集群模式安装部署的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1029667

相关文章

龙蜥操作系统Anolis OS-23.x安装配置图解教程(保姆级)

《龙蜥操作系统AnolisOS-23.x安装配置图解教程(保姆级)》:本文主要介绍了安装和配置AnolisOS23.2系统,包括分区、软件选择、设置root密码、网络配置、主机名设置和禁用SELinux的步骤,详细内容请阅读本文,希望能对你有所帮助... ‌AnolisOS‌是由阿里云推出的开源操作系统,旨

k8s部署MongDB全过程

《k8s部署MongDB全过程》文章介绍了如何在Kubernetes集群中部署MongoDB,包括环境准备、创建Secret、创建服务和Deployment,并通过Robo3T工具测试连接... 目录一、环境准备1.1 环境说明1.2 创建 namespace1.3 创建mongdb账号/密码二、创建Sec

Java中的Opencv简介与开发环境部署方法

《Java中的Opencv简介与开发环境部署方法》OpenCV是一个开源的计算机视觉和图像处理库,提供了丰富的图像处理算法和工具,它支持多种图像处理和计算机视觉算法,可以用于物体识别与跟踪、图像分割与... 目录1.Opencv简介Opencv的应用2.Java使用OpenCV进行图像操作opencv安装j

Ubuntu系统怎么安装Warp? 新一代AI 终端神器安装使用方法

《Ubuntu系统怎么安装Warp?新一代AI终端神器安装使用方法》Warp是一款使用Rust开发的现代化AI终端工具,该怎么再Ubuntu系统中安装使用呢?下面我们就来看看详细教程... Warp Terminal 是一款使用 Rust 开发的现代化「AI 终端」工具。最初它只支持 MACOS,但在 20

mysql-8.0.30压缩包版安装和配置MySQL环境过程

《mysql-8.0.30压缩包版安装和配置MySQL环境过程》该文章介绍了如何在Windows系统中下载、安装和配置MySQL数据库,包括下载地址、解压文件、创建和配置my.ini文件、设置环境变量... 目录压缩包安装配置下载配置环境变量下载和初始化总结压缩包安装配置下载下载地址:https://d

将Python应用部署到生产环境的小技巧分享

《将Python应用部署到生产环境的小技巧分享》文章主要讲述了在将Python应用程序部署到生产环境之前,需要进行的准备工作和最佳实践,包括心态调整、代码审查、测试覆盖率提升、配置文件优化、日志记录完... 目录部署前夜:从开发到生产的心理准备与检查清单环境搭建:打造稳固的应用运行平台自动化流水线:让部署像

LinuxMint怎么安装? Linux Mint22下载安装图文教程

《LinuxMint怎么安装?LinuxMint22下载安装图文教程》LinuxMint22发布以后,有很多新功能,很多朋友想要下载并安装,该怎么操作呢?下面我们就来看看详细安装指南... linux Mint 是一款基于 Ubuntu 的流行发行版,凭借其现代、精致、易于使用的特性,深受小伙伴们所喜爱。对

Python项目打包部署到服务器的实现

《Python项目打包部署到服务器的实现》本文主要介绍了PyCharm和Ubuntu服务器部署Python项目,包括打包、上传、安装和设置自启动服务的步骤,具有一定的参考价值,感兴趣的可以了解一下... 目录一、准备工作二、项目打包三、部署到服务器四、设置服务自启动一、准备工作开发环境:本文以PyChar

Linux(Centos7)安装Mysql/Redis/MinIO方式

《Linux(Centos7)安装Mysql/Redis/MinIO方式》文章总结:介绍了如何安装MySQL和Redis,以及如何配置它们为开机自启,还详细讲解了如何安装MinIO,包括配置Syste... 目录安装mysql安装Redis安装MinIO总结安装Mysql安装Redis搜索Red

python安装完成后可以进行的后续步骤和注意事项小结

《python安装完成后可以进行的后续步骤和注意事项小结》本文详细介绍了安装Python3后的后续步骤,包括验证安装、配置环境、安装包、创建和运行脚本,以及使用虚拟环境,还强调了注意事项,如系统更新、... 目录验证安装配置环境(可选)安装python包创建和运行Python脚本虚拟环境(可选)注意事项安装