百度360必应搜狗淘宝本站头条
mysql distinctjenkins官网mysql 子查询vba splitpython list sort
当前位置：网站首页 > IT技术 > 正文

Storm 集群搭建及编写WordCount

wptr33 2025-03-04 14:19 19 浏览

准备工作

1. 下载zookeeper-3.4.7

2. 下载Storm apache-storm-0.9.3

3. 安装JDK 1.7

注：

Storm0.9之前的版本，底层通讯用的是ZeroMQ，所以在安装0.9之前的版本需要安装0MQ，但是在0.9之后 我们直接安装就可以了。

因为在storm被移交到apache之后，这块用java的NIO矿建Netty代替了。

首先建立hadoop用户，我把和大数据相关的框架和软件都放在hadoop用户中。

安装ZK

1. 登陆到10.10.113.41并解压tar包

2. 建立zookeeper的data目录，
/home/hadoop/zookeeper/data

mkdir -p /home/hadoop/zookeeper/data

3. 建立zk集群的myid文件 （单机版可以跳过该步）

cd /home/hadoop/zookeeper/data

echo 1 > myid

4. 拷贝zookeeper的conf/zoo_sample.cfg并重命名为zoo.cfg，修改如下：

dataDir=/home/hadoop/zookeeper/data

server.1=10.10.113.41:2888:3888

server.2=10.10.113.42:2888:3888

server.3=10.10.113.43:2888:3888

dataDir是配置zk的数据目录的

server.A=B:C:D是集群zk使用的。如果你只想用单个zk，可以不配置。

A - 是一个数字，表示这是第几号服务器。与/var/tmp/zkdata下的myid文件内容一致

B - 是该服务器的IP地址

C - 表示该服务器与集群中的Leader服务器交换信息的端口

D - 表示如果万一集群中的Leader服务器挂了，需要各服务器重新选举时所用的通讯端口

5. (Optional)将zk的bin目录路径加入环境变量

修改/etc/profile文件，在尾部添加如下：

#zookeeper

export ZOOKEEPER==/home/hadoop/zookeeper

PATH=$PATH:$ZOOKEEPER/bin

6. 启动zk

zkServer.sh start

在剩下两台机器重复以上步骤，注意myid要对应

6.查看zk的运行状态

zkServer.sh status

安装Storm

1. 解压tar包并赋予执行权限

2. 将Storm的bin目录加入系统路径

修改/etc/profile文件，在尾部加入如下：

PATH=$PATH:/home/hadoop/storm

使其生效

3. 创建一个Storm的本地数据目录

mkdir -p /home/hadoop/storm/data

以上步骤在Storm的集群上的其他机器上重复执行，然后进行配置：

a. 配置storm.yaml

修改storm的conf/storm.yaml文件如下：

storm.zookeeper.servers: #zk地址

- "10.10.113.41"

- "10.10.113.42"

- "10.10.113.43"

nimbus.host: "10.10.113.41" #master 节点地址

supervisor.slots.ports:

- 6700

- 6701

- 6702

- 6703

storm.local.dir: "/home/hadoop/storm/data" #数据存放地址

注意：

在每个配置项前面必须留有空格，否则会无法识别。

启动集群

1. 启动nimbus

在nimbus机器的Storm的bin目录下执行

nohup bin/storm nimbus >/dev/null 2>&1 & #启动主节点

nohup bin/storm ui >/dev/null 2>&1 & #启动stormUI

nohup bin/storm logviewer >/dev/null 2>&1 & #启动logviewer 功能

2. 启动supervisor

在supervisor机器的Storm的bin目录下执行，所有supervisor节点都使用如下命令

nohup bin/storm supervisor >/dev/null 2>&1 &

nohup bin/storm logviewer >/dev/null 2>&1 &

3. 检查

打开Storm UI 页面。
http://10.10.113.41:8080/index.html

默认是启在8080端口上，如果你想改成其他的，如8089，直接修改nimbus的storm.yaml文件，添加

ui.port=8089

部署程序

1. 这里我使用 Intellij IDEA + maven来开发一个wordcount的Demo

2. 添加maven依赖

org.apache.storm

storm-core

0.9.3<

1. 这里我使用 Intellij IDEA + maven来开发一个wordcount的Demo部署程序

2. 添加maven依赖

org.apache.storm

storm-core

0.9.3

3. 新建项目，编写程序

package cn.oraclers.storm;

import backtype.storm.Config;

import backtype.storm.StormSubmitter;

import backtype.storm.generated.AlreadyAliveException;

import backtype.storm.generated.InvalidTopologyException;

import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.TopologyBuilder;

import backtype.storm.topology.base.BaseRichBolt;

import backtype.storm.topology.base.BaseRichSpout;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Tuple;

import backtype.storm.tuple.Values;

import backtype.storm.utils.Utils;

import java.util.HashMap;

import java.util.Map;

import java.util.Random;

public class WordCount {

public static class SpoutSource extends BaseRichSpout {

Map map;

TopologyContext topologyContext;

SpoutOutputCollector spoutOutputCollector;

Random random;

@Override

public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {

map = map;

topologyContext = topologyContext;

spoutOutputCollector = spoutOutputCollector;

random = random;

}

String[] sentences = new String[]{ "the cow jumped over the moon", "an apple a day keeps the doctor away",

"four score and seven years ago", "snow white and the seven dwarfs", "i am at two with nature" };

@Override

public void nextTuple() {

Utils.sleep(1000);

for (String sentence:sentences){

spoutOutputCollector.emit(new Values(sentence));

}

}

@Override

public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

outputFieldsDeclarer.declare(new Fields("sentence"));

}

}

public static class SplitBoltSource extends BaseRichBolt{

Map map;

TopologyContext topologyContext;

OutputCollector outputCollector;

@Override

public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

map = map;

topologyContext = topologyContext;

outputCollector = outputCollector;

}

@Override

public void execute(Tuple tuple) {

String sentence = tuple.getStringByField("sentence");

String[] words = sentence.split(" ");

for (String word:words){

this.outputCollector.emit(new Values(word));

}

}

@Override

public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

outputFieldsDeclarer.declare(new Fields("word"));

}

}

public static class SumBoltSource extends BaseRichBolt{

Map map;

TopologyContext topologyContext;

OutputCollector outputCollector;

@Override

public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

this.map = map;

this.topologyContext = topologyContext;

this.outputCollector = outputCollector;

}

Map mapCount = new HashMap();

@Override

public void execute(Tuple tuple) {

String word = tuple.getStringByField("word");

Integer count = mapCount.get(word);

if(count == null){

count=0;

}

count++;

mapCount.put(word,count);

outputCollector.emit(new Values(word,count));

}

@Override

public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

outputFieldsDeclarer.declare(new Fields("word", "count"));

}

}

public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("data_source",new SpoutSource());

builder.setBolt("bolt_split",new SplitBoltSource()).shuffleGrouping("data_source");

builder.setBolt("bolt_sum",new SplitBoltSource()).fieldsGrouping("bolt_split",new Fields("word"));

try {

Config stormConf = new Config();

stormConf.setDebug(true);

StormSubmitter.submitTopology("Clustertopology", stormConf,builder.createTopology());

} catch (AlreadyAliveException e) {

e.printStackTrace();

} catch (InvalidTopologyException e) {

e.printStackTrace();

}

}

}

4. 打包部署topology

./storm jar storm jar sd-1.0-SNAPSHOT.jar cn.oraclers.storm.WordCount

5. 查看结果

两种方式，

a. 查看StormUI

注意：一定不要使用IE内核的浏览器，否则看不到Topology Summary 下面的东西！！！

b. storm的bin目录下运行

Topology_name Status Num_tasks Num_workers Uptime_secs

-------------------------------------------------------------------

test ACTIVE 28 3 5254

Clustertopology ACTIVE 4 1 83

mytopo ACTIVE 6 3 555

6. 关闭topology

a. StormUI上面点选要关闭的topology，如test，然后在新页面的Topology actions中选kill

b. 运行./storm kill test

相关推荐

oracle数据导入导出_oracle数据导入导出工具

关于oracle的数据导入导出，这个功能的使用场景，一般是换服务环境，把原先的oracle数据导入到另外一台oracle数据库，或者导出备份使用。只不过oracle的导入导出命令不好记忆，稍稍有点复杂...

继续学习Python中的while true/break语句

上次讲到if语句的用法，大家在微信公众号问了小编很多问题，那么小编在这几种解决一下，1.else和elif是子模块，不能单独使用2.一个if语句中可以包括很多个elif语句，但结尾只能有一个else解...

python continue和break的区别_python中break语句和continue语句的区别

python中循环语句经常会使用continue和break，那么这2者的区别是？continue是跳出本次循环，进行下一次循环；break是跳出整个循环；例如：...

简单学Python——关键字6——break和continue

Python退出循环，有break语句和continue语句两种实现方式。break语句和continue语句的区别：break语句作用是终止循环。continue语句作用是跳出本轮循环，继续下一次循...

2-1，0基础学Python之 break退出循环、 continue继续循环 多重循

用for循环或者while循环时，如果要在循环体内直接退出循环，可以使用break语句。比如计算1至100的整数和，我们用while来实现：sum=0x=1whileTrue...

Python 中 break 和 continue 傻傻分不清

大家好啊，我是大田。今天分享一下break和continue在代码中的执行效果是什么，进一步区分出二者的区别。一、continue例1：当小明3岁时不打印年龄，其余年龄正常循环打印。可以看...

python中的流程控制语句:continue、break 和 return使用方法

Python中，continue、break和return是控制流程的关键语句，用于在循环或函数中提前退出或跳过某些操作。它们的用途和区别如下：1.continue（跳过当前循环的剩余部分，进...

L017:continue和break - 教程文案

continue和break在Python中，continue和break是用于控制循环（如for和while）执行流程的关键字，它们的作用如下：1.continue：跳过当前迭代，...

作为前端开发者，你都经历过怎样的面试?

已经裸辞1个月了，最近开始投简历找工作，遇到各种各样的面试，今天分享一下。其实在职的时候也做过面试官，面试官时，感觉自己问的问题很难区分候选人的能力，最好的办法就是看看候选人的github上的代码仓库...

面试被问 const 是否不可变?这样回答才显功底

作为前端开发者，我在学习ES6特性时，总被const的"善变"搞得一头雾水——为什么用const声明的数组还能push元素？为什么基本类型赋值就会报错？直到翻遍MDN文档、对着内存图反...

2023金九银十必看前端面试题!2w字精品!

导文2023金九银十必看前端面试题！金九银十黄金期来了想要跳槽的小伙伴快来看啊CSS1.请解释CSS的盒模型是什么，并描述其组成部分。答案：CSS的盒模型是用于布局和定位元素的概念。它由内容区域...

前端面试总结_前端面试题整理

记得当时大二的时候，看到实验室的学长学姐忙于各种春招，有些收获了大厂offer，有些还在苦苦面试，其实那时候的心里还蛮忐忑的，不知道自己大三的时候会是什么样的一个水平，所以从19年的寒假放完，大二下学...

由浅入深，66条JavaScript面试知识点(七)

作者：JakeZhang转发链接：https://juejin.im/post/5ef8377f6fb9a07e693a6061目录由浅入深，66条JavaScript面试知识点（一）由浅入深，66...

2024前端面试真题之—VUE篇_前端面试题vue2020及答案

添加图片注释，不超过140字（可选）1.vue的生命周期有哪些及每个生命周期做了什么？beforeCreate是newVue()之后触发的第一个钩子，在当前阶段data、methods、com...

今年最常见的前端面试题，你会做几道?

在面试或招聘前端开发人员时，期望、现实和需求之间总是存在着巨大差距。面试其实是一个交流想法的地方，挑战人们的思考方式，并客观地分析给定的问题。可以通过面试了解人们如何做出决策，了解一个人对技术和解决问...

一周热门
最近发表
标签列表
赣ICP备2024040476号-12本站文章收集于网络以及网友投稿，其版权归原创者所有，如有侵害到你的权益，或有不妥之处，请联系删除。