一个最简单的JStorm例子
2024-08-24 10:59:21
最简单的JStorm例子分为以下几个步骤:
1、生成Topology
Map conf = new HashMp();
//topology所有自定义的配置均放入这个Map TopologyBuilder builder = new TopologyBuilder();
//创建topology的生成器 int spoutParal = get("spout.parallel", 1);
//获取spout的并发设置 SpoutDeclarer spout = builder.setSpout(SequenceTopologyDef.SEQUENCE_SPOUT_NAME,
new SequenceSpout(), spoutParal);
//创建Spout, 其中new SequenceSpout() 为真正spout对象,SequenceTopologyDef.SEQUENCE_SPOUT_NAME 为spout的名字,注意名字中不要含有空格 int boltParal = get("bolt.parallel", 1);
//获取bolt的并发设置 BoltDeclarer totalBolt = builder.setBolt(SequenceTopologyDef.TOTAL_BOLT_NAME, new TotalCount(),
boltParal).shuffleGrouping(SequenceTopologyDef.SEQUENCE_SPOUT_NAME);
//创建bolt, SequenceTopologyDef.TOTAL_BOLT_NAME 为bolt名字,TotalCount 为bolt对象,boltParal为bolt并发数,
//shuffleGrouping(SequenceTopologyDef.SEQUENCE_SPOUT_NAME),
//表示接收SequenceTopologyDef.SEQUENCE_SPOUT_NAME的数据,并且以shuffle方式,
//即每个spout随机轮询发送tuple到下一级bolt中 int ackerParal = get("acker.parallel", 1);
Config.setNumAckers(conf, ackerParal);
//设置表示acker的并发数 int workerNum = get("worker.num", 10);
conf.put(Config.TOPOLOGY_WORKERS, workerNum);
//表示整个topology将使用几个worker conf.put(Config.STORM_CLUSTER_MODE, "distributed");
//设置topolog模式为分布式,这样topology就可以放到JStorm集群上运行 StormSubmitter.submitTopology(streamName, conf,
builder.createTopology());
//提交topology
2、IRichSpout
IRichSpout 为最简单的Spout接口
IRichSpout{ @Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
} @Override
public void close() {
} @Override
public void activate() {
} @Override
public void deactivate() {
} @Override
public void nextTuple() {
} @Override
public void ack(Object msgId) {
} @Override
public void fail(Object msgId) {
} @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
} @Override
public Map<String, Object> getComponentConfiguration() {
return null;
}
其中注意:
- spout对象必须是继承Serializable, 因此要求spout内所有数据结构必须是可序列化的
- spout可以有构造函数,但构造函数只执行一次,是在提交任务时,创建spout对象,因此在task分配到具体worker之前的初始化工作可以在此处完成,一旦完成,初始化的内容将携带到每一个task内(因为提交任务时将spout序列化到文件中去,在worker起来时再将spout从文件中反序列化出来)。
- open是当task起来后执行的初始化动作
- close是当task被shutdown后执行的动作
- activate 是当task被激活时,触发的动作
- deactivate 是task被deactive时,触发的动作
- nextTuple 是spout实现核心, nextuple完成自己的逻辑,即每一次取消息后,用collector 将消息emit出去。
- ack, 当spout收到一条ack消息时,触发的动作,详情可以参考 ack机制
- fail, 当spout收到一条fail消息时,触发的动作,详情可以参考 ack机制
- declareOutputFields, 定义spout发送数据,每个字段的含义
- getComponentConfiguration 获取本spout的component 配置
3、Bolt
IRichBolt { @Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
} @Override
public void execute(Tuple input) {
} @Override
public void cleanup() {
} @Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
} @Override
public Map<String, Object> getComponentConfiguration() {
return null;
} }
其中注意:
- bolt对象必须是继承Serializable, 因此要求spout内所有数据结构必须是可序列化的
- bolt可以有构造函数,但构造函数只执行一次,是在提交任务时,创建bolt对象,因此在task分配到具体worker之前的初始化工作可以在此处完成,一旦完成,初始化的内容将携带到每一个task内(因为提交任务时将bolt序列化到文件中去,在worker起来时再将bolt从文件中反序列化出来)。
- prepare是当task起来后执行的初始化动作
- cleanup是当task被shutdown后执行的动作
- execute是bolt实现核心, 完成自己的逻辑,即接受每一次取消息后,处理完,有可能用collector 将产生的新消息emit出去。 ** 在executor中,当程序处理一条消息时,需要执行collector.ack, 详情可以参考 ack机制 ** 在executor中,当程序无法处理一条消息时或出错时,需要执行collector.fail ,详情可以参考 ack机制
- declareOutputFields, 定义bolt发送数据,每个字段的含义
- getComponentConfiguration 获取本bolt的component 配置
4、编译
在Maven中配置
<dependency>
<groupId>com.alibaba.jstorm</groupId>
<artifactId>jstorm-client</artifactId>
<version>0.9.3.1</version>
<scope>provided</scope>
</dependency> <dependency>
<groupId>com.alibaba.jstorm</groupId>
<artifactId>jstorm-client-extension</artifactId>
<version>0.9.3.1</version>
<scope>provided</scope>
</dependency>
如果找不到jstorm-client和jstorm-client-extension包,可以自己下载jstorm源码进行编译,请参考 源码编译
打包时,需要将所有依赖打入到一个包中
<build>
<plugins> <plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass>storm.starter.SequenceTopology</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
</configuration>
</plugin>
</plugins>
</build>
5、提交jar
jstorm jar xxxxxx.jar com.alibaba.xxxx.xx parameter
- xxxx.jar 为打包后的jar
- com.alibaba.xxxx.xx 为入口类,即提交任务的类
- parameter即为提交参数
最新文章
- MySQL入门01-MySQL源码安装
- 优化Google字体 全面加速WordPress
- 【Android】你应该知道的调试神器----adb
- php获取从百度搜索进入网站的关键词
- SQL Server调优系列进阶篇 - 查询语句运行几个指标值监测
- 阿里云Linux启动tomcat并能外网访问
- 点击页面其它地方隐藏该div的方法
- Vue01 Vue介绍、Vue使用、Vue实例的创建、数据绑定、Vue实例的生命周期、差值与表达式、指令与事件、语法糖
- Ubuntu 的 bash和dash的区别
- 了解真实的rem手机屏幕适配
- maven项目中利用jmeter-maven-plugin插件直接执行jmeter jmx脚本
- php简单使用shmop函数创建共享内存减少服务器负载
- java爬虫进阶 —— ip池使用,iframe嵌套,异步访问破解
- for循环输出菱形
- ubuntu16.04下安装pcl点云库
- sdfsdfsdf
- 死磕salt系列-salt grains pillar 配置
- C++之函数模板
- 抽样分布(2) t分布
- 注解失效,@SpringBootApplication 失效,引入包失效
热门文章
- Linux命令-网络命令:lastlog
- SQL Server Profiler	跟踪sql小技巧
- pandas所占内存释放
- homebrew osx下面最优秀的包管理工具
- Unity3D发布安卓报错permisson denied的解决
- 4.lists(双向链表)
- html-文本处理
- [转]C++11 标准新特性:Defaulted 和 Deleted 函数
- Ream的入门使用
- (译)Getting Started——1.3.3 Working with Foundation(使用Foundation框架)