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

Storm 集群搭建及编写WordCount

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

准备工作

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

相关推荐

redis的八种使用场景

前言:redis是我们工作开发中，经常要打交道的，下面对redis的使用场景做总结介绍也是对redis举报的功能做梳理。缓存Redis最常见的用途是作为缓存，用于加速应用程序的响应速度。...

基于Redis的3种分布式ID生成策略

在分布式系统设计中，全局唯一ID是一个基础而关键的组件。随着业务规模扩大和系统架构向微服务演进，传统的单机自增ID已无法满足需求。高并发、高可用的分布式ID生成方案成为构建可靠分布式系统的必要条件。R...

基于OpenWrt系统路由器的模式切换与网页设计

摘要：目前商用WiFi路由器已应用到多个领域，商家通过给用户提供一个稳定免费WiFi热点达到吸引客户、提升服务的目标。传统路由器自带的Luci界面提供了工厂模式的Web界面，用户可通过该界面配置路...

这篇文章教你看明白 nginx-ingress 控制器

主机nginx一般nginx做主机反向代理（网关）有以下配置...

如何用redis实现注册中心

一句话总结使用Redis实现注册中心：服务注册...

爱可可老师24小时热门分享(2020.5.10)

No1.看自己以前写的代码是种什么体验？No2.DooM-chip!国外网友SylvainLefebvre自制的无CPU、无操作码、无指令计数器...No3.我认为CS学位可以更好，如...

Apportable:拯救程序员，IOS一秒变安卓

摘要：还在为了跨平台使用cocos2d-x吗,拯救objc程序员的奇葩来了,ApportableSDK:FreeAndroidsupportforcocos2d-iPhone。App...

JAVA实现超买超卖方案汇总，那个最适合你，一篇文章彻底讲透

以下是几种Java实现超买超卖问题的核心解决方案及代码示例，针对高并发场景下的库存扣减问题：方案一：Redis原子操作+Lua脚本（推荐）//使用Redis+Lua保证原子性publicbo...

3月26日更新 快速施法自动施法可独立设置

2016年3月26日DOTA2有一个79.6MB的更新主要是针对自动施法和快速施法的调整本来内容不多不少朋友都有自动施法和快速施法的困扰英文更新日志一些视觉BUG修复就不翻译了主要翻译自动施...

Redis 是如何提供服务的

在刚刚接触Redis的时候，最想要知道的是一个’setnameJhon’命令到达Redis服务器的时候，它是如何返回’OK’的？里面命令处理的流程如何，具体细节怎么样？你一定有问过自己...

lua _G、_VERSION使用

到这里我们已经把lua基础库中的函数介绍完了，除了函数外基础库中还有两个常量，一个是_G，另一个是_VERSION。_G是基础库本身，指向自己，这个变量很有意思，可以无限引用自己，最后得到的还是自己，...

China&#39;s top diplomat to chair third China-Pacific Island countries foreign ministers&#39; meeting

BEIJING,May21(Xinhua)--ChineseForeignMinisterWangYi,alsoamemberofthePoliticalBureau...

移动工作交流工具Lua推出Insights数据分析产品

Lua是一个适用于各种职业人士的移动交流平台，它在今天推出了一项叫做Insights的全新功能。Insights是一个数据平台，客户可以在上面实时看到员工之间的交流情况，并分析这些情况对公司发展的影响...

Redis 7新武器:用Redis Stack实现向量搜索的极限压测

当传统关系型数据库还在为向量相似度搜索的性能挣扎时，Redis7的RedisStack...

Nginx/OpenResty详解，Nginx Lua编程，重定向与内部子请求

重定向与内部子请求Nginx的rewrite指令不仅可以在Nginx内部的server、location之间进行跳转，还可以进行外部链接的重定向。通过ngx_lua模块的Lua函数除了能实现Nginx...

一周热门
最近发表
标签列表
赣ICP备2024040476号-12