storm完成wc
spout 从数据源取数据 继承 baserichspout
在 spout中 1 完成nextTuple() 将数据定义成元组发送 collector.emit(new Values(lines)) 2 完成declare函数 定义发送的数据的字段名称 declare.declare(new Fields(""))
bolt 对数据处理 可以继承baserichbolt 需要数据等待一起发送 继承 basebatchbolt 1 在 execute() 函数中 先根据字段名称接收spout/bolt发送的数据 2 完成数据处理 转换 不需要往下级发送数据 不用完成declare函数
Main函数 new TopologyBuilder 封装拓扑计算逻辑 builder.setSpout builder.setBolt 将自定义的spout bolt封装进去 builder.createTopology() 创建一个 拓扑的实例
drpc 远程调用方法
LinearDRPCTopologyBuilder 初始化方法名 或者 new Topologybuilder() 然后 new DRPCSpout("",drpc) 在远程 通过指定服务器位置 端口号3772 方法名称 完成远程调用
storm向HBase写结果 在最后一个 bolt中的execute中加入hbase的操作 处理 其他处理不变
acker 消息完整性的实现机制
在 spout端 完成 ack() fail() 函数 消息失败完成重发逻辑 在bolt发送带锚标记的collector.emit(input,new Values(" "))
bin/storm nimbus >> logs/nimbus.out 2>&1 &
bin/storm ui >> logs/ui.out 2>&1 &
bin/storm supervisor >> logs/supervisor.out 2>&1 &
bin/storm logviewer >> logs/logviewer.out 2>&1 &
bin/storm drpc >> logs/drpc.out 2>&1 &