本文主要介紹如何在Storm編程實現(xiàn)與Kafka的集成
一、實現(xiàn)模型
數(shù)據(jù)流程：
1、Kafka Producter生成topic1主題的消息
2、Storm中有個Topology，包含了KafkaSpout、SenqueceBolt、KafkaBolt三個組件。其中KafkaSpout訂閱了topic1主題消息，然后發(fā)送
給SenqueceBolt加工處理，最后數(shù)據(jù)由KafkaBolt生成topic2主題消息發(fā)送給Kafka
3、Kafka Consumer負(fù)責(zé)消費topic2主題的消息
二、Topology實現(xiàn)
1、創(chuàng)建maven工程，配置pom.xml
需要依賴storm-core、kafka_2.10、storm-kafka三個包
<dependencies> <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>0.9.2-incubating</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.10</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> </dependencies> <build> <plugins> <plugin> <artifactId>maven-assembly-plugin</artifactId> <version>2.4</version> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> </plugins> </build>
2、KafkaSpout
KafkaSpout是Storm中自帶的Spout，源碼在https://github.com/apache/incubator-storm/tree/master/external
使用KafkaSpout時需要子集實現(xiàn)Scheme接口，它主要負(fù)責(zé)從消息流中解析出需要的數(shù)據(jù)
public class MessageScheme implements Scheme { /* (non-Javadoc) * @see backtype.storm.spout.Scheme#deserialize(byte[]) */ public List<Object> 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"); } } 3、SenqueceBolt
SenqueceBolt實現(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")); } } 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é)點目錄和名字 SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "topic1", "/zkkafkaspout" , "kafkaspout");
// 配置KafkaBolt中的kafka.broker.properties Config conf = new Config(); Map<String, String> map = new HashMap<String, String>();
// 配置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<String, Integer>()).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(); } } }
三、測試驗證
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、運行Strom Topology
bin/storm jar storm-kafka-0.0.1-SNAPSHOT-jar-with-dependencies.jar StormKafkaTopo KafkaStorm
4、運行結(jié)果
posted on 2015-03-01 15:47
SIMONE 閱讀(4957)
評論(0) 編輯 收藏 所屬分類:
hadoop