問題導(dǎo)讀
1.如何編程實(shí)現(xiàn)Storm與Kafka集成？
2.Storm中Topology如何實(shí)現(xiàn)的？
3.如何驗(yàn)證集成效果？
一、實(shí)現(xiàn)模型
數(shù)據(jù)流程：
1、Kafka Producter生成topic1主題的消息
2、Storm中有個(gè)Topology，包含了KafkaSpout、SenqueceBolt、KafkaBolt三個(gè)組件。其中KafkaSpout訂閱了topic1主題消息，然后發(fā)送
給SenqueceBolt加工處理，最后數(shù)據(jù)由KafkaBolt生成topic2主題消息發(fā)送給Kafka
3、Kafka Consumer負(fù)責(zé)消費(fèi)topic2主題的消息
二、Topology實(shí)現(xiàn)
1、創(chuàng)建maven工程，配置pom.xml
需要依賴storm-core、kafka_2.10、storm-kafka三個(gè)包
org.apache.storm
storm-core
0.9.2-incubating
provided
org.apache.kafka
kafka_2.10
0.8.1.1
org.apache.zookeeper
zookeeper
log4j
log4j
org.apache.storm
storm-kafka
0.9.2-incubating
maven-assembly-plugin
2.4
jar-with-dependencies
make-assembly
package
single
復(fù)制代碼
2、KafkaSpout
KafkaSpout是Storm中自帶的Spout，源碼在https://github.com/apache/incubator-storm/tree/master/external
使用KafkaSpout時(shí)需要子集實(shí)現(xiàn)Scheme接口，它主要負(fù)責(zé)從消息流中解析出需要的數(shù)據(jù)
public class MessageScheme implements Scheme {
/* (non-Javadoc)
* @see backtype.storm.spout.Scheme#deserialize(byte[])
*/
public List deserialize(byte[] ser) {
try {
String msg = new String(ser, "UTF-8");
return new Values(msg);
} catch (UnsupportedEncodingException e) {
}
return null;
}
/* (non-Javadoc)
* @see backtype.storm.spout.Scheme#getOutputFields()
*/
public Fields getOutputFields() {
// TODO Auto-generated method stub
return new Fields("msg");
}
}
復(fù)制代碼
3、SenqueceBolt
SenqueceBolt實(shí)現(xiàn)很簡單，在接收的spout的消息前面加上“I‘m”
public class SenqueceBolt extends BaseBasicBolt{
/* (non-Javadoc)
* @see backtype.storm.topology.IBasicBolt#execute(backtype.storm.tuple.Tuple, backtype.storm.topology.BasicOutputCollector)
*/
public void execute(Tuple input, BasicOutputCollector collector) {
// TODO Auto-generated method stub
String word = (String) input.getValue(0);
String out = "I'm " + word +??"!";
System.out.println("out=" + out);
collector.emit(new Values(out));
}
/* (non-Javadoc)
* @see backtype.storm.topology.IComponent#declareOutputFields(backtype.storm.topology.OutputFieldsDeclarer)
*/
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("message"));
}
}
復(fù)制代碼
4、KafkaBolt
KafkaBolt是Storm中自帶的Bolt，負(fù)責(zé)向Kafka發(fā)送主題消息
5、Topology
public class StormKafkaTopo {
public static void main(String[] args) throws Exception {
// 配置Zookeeper地址
BrokerHosts brokerHosts = new ZkHosts("node04:2181,node05:2181,node06:2181");
// 配置Kafka訂閱的Topic，以及zookeeper中數(shù)據(jù)節(jié)點(diǎn)目錄和名字
SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "topic1", "/zkkafkaspout" , "kafkaspout");
// 配置KafkaBolt中的kafka.broker.properties
Config conf = new Config();
Map map = new HashMap();
// 配置Kafka broker地址
map.put("metadata.broker.list", "node04:9092");
// serializer.class為消息的序列化類
map.put("serializer.class", "kafka.serializer.StringEncoder");
conf.put("kafka.broker.properties", map);
// 配置KafkaBolt生成的topic
conf.put("topic", "topic2");
spoutConfig.scheme = new SchemeAsMultiScheme(new MessageScheme());
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new KafkaSpout(spoutConfig));
builder.setBolt("bolt", new SenqueceBolt()).shuffleGrouping("spout");
builder.setBolt("kafkabolt", new KafkaBolt()).shuffleGrouping("bolt");
if (args != null && args.length > 0) {
conf.setNumWorkers(3);
StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
} else {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("Topo", conf, builder.createTopology());
Utils.sleep(100000);
cluster.killTopology("Topo");
cluster.shutdown();
}
}
}
復(fù)制代碼
三、測試驗(yàn)證
1、使用Kafka client模擬Kafka Producter ，生成topic1主題
bin/kafka-console-producer.sh --broker-list node04:9092 --topic topic1
2、使用Kafka client模擬Kafka Consumer，訂閱topic2主題
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic topic2 --from-beginning
3、運(yùn)行Strom Topology
bin/storm jar storm-kafka-0.0.1-SNAPSHOT-jar-with-dependencies.jar??StormKafkaTopo KafkaStorm
4、運(yùn)行結(jié)果
原創(chuàng)文章，轉(zhuǎn)載請注明： 轉(zhuǎn)載自http://www.cnblogs.com/tovin/p/3974417.html
public class StormKafkaTopo {
public static void main(String[] args) throws Exception {
// 配置Zookeeper地址
BrokerHosts brokerHosts = new ZkHosts("storm1:2181,storm2:2181,storm3:2181");
// 配置Kafka訂閱的Topic，以及zookeeper中數(shù)據(jù)節(jié)點(diǎn)目錄和名字
SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "topic1", "/zkkafkaspout" , "kafkaspout");
// 配置KafkaBolt中的kafka.broker.properties
Config conf = new Config();
Map map = new HashMap();
// 配置Kafka broker地址
map.put("metadata.broker.list", "storm3:9092");
// serializer.class為消息的序列化類
map.put("serializer.class", "kafka.serializer.StringEncoder");
conf.put("kafka.broker.properties", map);
// 配置KafkaBolt生成的topic
conf.put("topic", "topic2");
spoutConfig.scheme = new SchemeAsMultiScheme(new MessageScheme());
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new KafkaSpout(spoutConfig));
builder.setBolt("bolt", new SenqueceBolt()).shuffleGrouping("spout");
builder.setBolt("kafkabolt", new KafkaBolt()).shuffleGrouping("bolt");
if (args != null && args.length > 0) {
conf.setNumWorkers(3);
StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
} else {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("Topo", conf, builder.createTopology());
Utils.sleep(100000);
cluster.killTopology("Topo");
cluster.shutdown();
}
}
}
