1.環(huán)境介紹

?如圖所示，NODEJS做為數(shù)據(jù)源的的產(chǎn)生者產(chǎn)生消息，發(fā)到Kafka隊(duì)列，然后參見紅線，表示本地開發(fā)的環(huán)境下數(shù)據(jù)的流向（本地開發(fā)時(shí)，storm topology運(yùn)行在本地模式）

2.搭建環(huán)境,我采用的是eclipse+maven

1.建立一個(gè)maven工程， 然后將pom文件修改如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.h3c.storm</groupId><artifactId>storm-samples</artifactId><packaging>jar</packaging><version>1.0-SNAPSHOT</version><name>storm-kafka-test</name><url>http://maven.apache.org</url><dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>3.8.1</version><scope>test</scope></dependency><dependency> <groupId>jdk.tools</groupId> <artifactId>jdk.tools</artifactId> <version>1.7</version> <scope>system</scope> <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath> </dependency> <dependency><groupId>org.apache.storm</groupId><artifactId>storm-core</artifactId><version>0.10.0</version><!-- keep storm out of the jar-with-dependencies --><scope>provided</scope></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka_2.9.2</artifactId><version>0.8.1.1</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion></exclusions></dependency><dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-kafka</artifactId> <version>0.9.2-incubating</version> </dependency><dependency><groupId>org.apache.storm</groupId><artifactId>storm-hbase</artifactId><version>0.10.0</version></dependency> </dependencies> </project> View Code

?

2.nodeJS發(fā)消息的示例代碼，當(dāng)然，首先要手動(dòng)在kafka里新建一個(gè)topic對應(yīng)代碼里的topic，我這里創(chuàng)建的topic是"historyclients"

var kafka = require('kafka-node'); var Producer = kafka.Producer; var KeyedMessage = kafka.KeyedMessage; var conf = '172.27.8.111:2181,172.27.8.112:2181,172.27.8.119:2181'; var client = new kafka.Client(conf); var producer = new Producer(client);var clientOnlineInfo ={"clientMAC":"0000-0000-0002","acSN":"210235A1AMB159000008","onLineTime":"2016-06-27 10:00:00"};var clientOnlineInfoStr = JSON.stringify(clientOnlineInfo);var msg = [{ topic: 'historyclients', messages: clientOnlineInfoStr, partition: 0 } ];producer.on('ready', function () {producer.send(msg, function (err, data) {console.log("done!")console.log(data);}); });producer.on('error', function (err) {console.error(err); }); View Code

?

3.spout代碼

package com.h3c.storm;import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties;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 kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector;public class KafkaSpout extends BaseRichSpout{private SpoutOutputCollector collector;private ConsumerConnector consumer; private String topic; Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap;private static ConsumerConfig createConsumerConfig() { Properties props = new Properties(); props.put("zookeeper.connect", "172.27.8.111:2181,172.27.8.112:2181,172.27.8.119:2181"); props.put("group.id", "group1"); props.put("zookeeper.session.timeout.ms", "40000"); props.put("zookeeper.sync.time.ms", "200"); props.put("auto.commit.interval.ms", "1000"); return new ConsumerConfig(props); } @Overridepublic void open(Map conf, TopologyContext context,SpoutOutputCollector collector) {System.err.println("open!!!!!!!!!!!!!!!");this.collector = collector;/* create consumer */this.topic = "historyclients";this.consumer = kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig()); /* topic HashMap,which means the map can include multiple topics */Map<String, Integer> topicCountMap = new HashMap<String, Integer>();topicCountMap.put(topic, new Integer(1)); this.consumerMap = consumer.createMessageStreams(topicCountMap); }@Overridepublic void nextTuple() {KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0); ConsumerIterator<byte[], byte[]> it = stream.iterator(); String toSay = "";while (it.hasNext()) {toSay = new String(it.next().message());System.err.println("receive：" + toSay); this.collector.emit(new Values(toSay));} }@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("clientInfo"));} } View Code

4.storm-hbase API 中要求實(shí)現(xiàn)的mapper代碼

package com.h3c.storm;import org.apache.storm.hbase.bolt.mapper.HBaseMapper; import org.apache.storm.hbase.common.ColumnList;import backtype.storm.tuple.Tuple;public class MyHBaseMapper implements HBaseMapper {public ColumnList columns(Tuple tuple) {ColumnList cols = new ColumnList();//參數(shù)依次是列族名，列名，值cols.addColumn("f1".getBytes(), "colMAC".getBytes(), tuple.getStringByField("clientInfo").getBytes());//System.err.println("BOLT + " + tuple.getStringByField("clientInfo"));//cols.addColumn("f1".getBytes(), "hhhhhhh".getBytes(), "0000-0000-0001".getBytes());//System.err.println("BOLT + " + tuple.getStringByField("clientInfo"));return cols;}public byte[] rowKey(Tuple tuple) {//return tuple.getStringByField("clientInfo").getBytes();return "newRowKey".getBytes(); }} Mapper

5.topology代碼

package com.h3c.storm;import java.util.Map; import java.util.Random;import org.apache.storm.hbase.bolt.HBaseBolt; import org.apache.storm.hbase.bolt.mapper.HBaseMapper;import java.util.HashMap; import java.util.List; import java.util.Properties; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.StormSubmitter; import backtype.storm.spout.SpoutOutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.TopologyBuilder; import backtype.storm.topology.base.*; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.Values; import backtype.storm.utils.Utils; import kafka.consumer.Consumer; import kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.ConsumerTimeoutException; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import kafka.message.MessageAndMetadata;public class PersistTopology {private static final String KAFKA_SPOUT = "KAFKA_SPOUT";private static final String HBASE_BOLT = "HBASE_BOLT";public static void main(String[] args) throws Exception {/* define spout */KafkaSpout kafkaSpout = new KafkaSpout();System.setProperty("hadoop.home.dir", "E:\\eclipse\\");/* define HBASE Bolt */HBaseMapper mapper = new MyHBaseMapper();HBaseBolt hbaseBolt = new HBaseBolt("historyclients", mapper).withConfigKey("hbase.conf");/* define topology*/TopologyBuilder builder = new TopologyBuilder();builder.setSpout(KAFKA_SPOUT, kafkaSpout);builder.setBolt(HBASE_BOLT, hbaseBolt).shuffleGrouping(KAFKA_SPOUT);Config conf = new Config();conf.setDebug(true);Map<String, Object> hbConf = new HashMap<String, Object>(); // if(args.length > 0){ // hbConf.put("hbase.rootdir", args[0]); // }//hbConf.put("hbase.rootdir", "hdfs://172.27.8.111:8020/apps/hbase/data");conf.put("hbase.conf", hbConf);if (args != null && args.length > 0) {conf.setNumWorkers(3);StormSubmitter.submitTopology(args[0], conf, builder.createTopology());} else {LocalCluster cluster = new LocalCluster();cluster.submitTopology("test", conf, builder.createTopology());Utils.sleep(600000);cluster.killTopology("test");cluster.shutdown();}} } View Code

6.需要從集群中取中hbase-site.xml這個(gè)文件，加到項(xiàng)目里，在buildpath中可設(shè)置

7.在C:\Windows\System32\drivers\etc下把hosts文件加上到集群的IP與域名的映射

172.27.8.111 node1.hde.h3c.com node1
172.27.8.112 node2.hde.h3c.com node2
172.27.8.119 node3.hde.h3c.com node3

8. 出現(xiàn)java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.的解決辦法

網(wǎng)上下載winutils.exe這個(gè)文件，找一個(gè)地方放好，比如我放在E:\eclipse\bin 下面，前面一定要有個(gè)“bin”

然后在代碼里加上這句即可

System.setProperty("hadoop.home.dir", "E:\\eclipse\\");

?

參考文章?

http://www.tuicool.com/articles/r6ZZBjU

轉(zhuǎn)載于:https://www.cnblogs.com/zhengchunhao/p/5630052.html

總結(jié)

以上是生活随笔為你收集整理的nodejs+kafka+storm+hbase 开发的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。