Kafka 与 Storm 整合

原创编写： 王宇 
2016-10-27

整合Storm的概念

• 整合场景

  A spout is a source of streams. For example, a spout may read tuples off a Kafka Topic and emit them as a stream. A bolt consumes input streams, process and possibly emits new streams. Bolts can do anything from running functions, filtering tuples, do streaming aggregations, streaming joins, talk to databases, and more. Each node in a Storm topology executes in parallel. A topology runs indefinitely until you terminate it. Storm will automatically reassign any failed tasks. Additionally, Storm guarantees that there will be no data loss, even if the machines go down and messages are dropped.

• BrokerHosts - ZkHosts & StaticHosts 
  BrokerHosts is an interface and ZkHosts and StaticHosts are its two main implementations. ZkHosts is used to track the Kafka brokers dynamically by maintaining the details in ZooKeeper, while StaticHosts is used to manually / statically set the Kafka brokers and its details. ZkHosts is the simple and fast way to access the Kafka broker

• KafkaConfig API 
  This API is used to define configuration settings for the Kafka cluster. The signature of Kafka Con-fig is defined as follows 
  public KafkaConfig(BrokerHosts hosts, string topic)

  Hosts − The BrokerHosts can be ZkHosts / StaticHosts. 
  Topic − topic name

• SpoutConfig API 
  Spoutconfig is an extension of KafkaConfig that supports additional ZooKeeper information. 
  public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)

  • Hosts − The BrokerHosts can be any implementation of BrokerHosts interface
  • Topic − topic name.
  • zkRoot − ZooKeeper root path.
  • id − The spout stores the state of the offsets its consumed in Zookeeper. The id should uniquely identify your spout.

• SchemeAsMultiScheme (调度)????

• KafkaSpout API 
  KafkaSpout is our spout implementation, which will integrate with Storm. It fetches the messages from kafka topic and emits it into Storm ecosystem as tuples. KafkaSpout get its config-uration details from SpoutConfig.

1. // ZooKeeper connection string
2. BrokerHosts hosts =newZkHosts(zkConnString);
4. //Creating SpoutConfig Object
5. SpoutConfig spoutConfig =newSpoutConfig(hosts,
6. topicName,"/"+ topicName UUID.randomUUID().toString());
8. //convert the ByteBuffer to String.
9. spoutConfig.scheme =newSchemeAsMultiScheme(newStringScheme());
11. //Assign SpoutConfig to KafkaSpout.
12. KafkaSpout kafkaSpout =newKafkaSpout(spoutConfig);

创建 Bolt

• Bolt 接口定义参考《Storm学习笔记》
• 分割统计单词个数的Bolt例子

  • SplitBolt.java

1. import java.util.Map;
3. import org.apache.storm.tuple.Tuple;
4. import org.apache.storm.tuple.Fields;
5. import org.apache.storm.tuple.Values;
7. import org.apache.storm.task.OutputCollector;
8. import org.apache.storm.topology.OutputFieldsDeclarer;
9. import org.apache.storm.topology.IRichBolt;
10. import org.apache.storm.task.TopologyContext;
12. publicclassSplitBoltimplementsIRichBolt{
13. privateOutputCollector collector;
15. @Override
16. publicvoid prepare(Map stormConf,TopologyContext context,
17. OutputCollector collector){
18. this.collector = collector;
19. }
21. @Override
22. publicvoid execute(Tuple input){
23. String sentence = input.getString(0);
24. String[] words = sentence.split(" ");
26. for(String word: words){
27. word = word.trim();
29. if(!word.isEmpty()){
30. word = word.toLowerCase();
31. collector.emit(newValues(word));
32. }
34. }
36. collector.ack(input);
37. }
39. @Override
40. publicvoid declareOutputFields(OutputFieldsDeclarer declarer){
41. declarer.declare(newFields("word"));
42. }
44. @Override
45. publicvoid cleanup(){}
47. @Override
48. publicMap<String,Object> getComponentConfiguration(){
49. returnnull;
50. }
52. }

• CountBolt.java

1. import java.util.Map;
2. import java.util.HashMap;
4. import org.apache.storm.tuple.Tuple;
5. import org.apache.storm.task.OutputCollector;
6. import org.apache.storm.topology.OutputFieldsDeclarer;
7. import org.apache.storm.topology.IRichBolt;
8. import org.apache.storm.task.TopologyContext;
10. publicclassCountBoltimplementsIRichBolt{
11. Map<String,Integer> counters;
12. privateOutputCollector collector;
14. @Override
15. publicvoid prepare(Map stormConf,TopologyContext context,
16. OutputCollector collector){
17. this.counters =newHashMap<String,Integer>();
18. this.collector = collector;
19. }
21. @Override
22. publicvoid execute(Tuple input){
23. String str = input.getString(0);
25. if(!counters.containsKey(str)){
26. counters.put(str,1);
27. }else{
28. Integer c = counters.get(str)+1;
29. counters.put(str, c);
30. }
32. collector.ack(input);
33. }
35. @Override
36. publicvoid cleanup(){
37. for(Map.Entry<String,Integer> entry:counters.entrySet()){
38. System.out.println(entry.getKey()+" : "+ entry.getValue());
39. }
40. }
42. @Override
43. publicvoid declareOutputFields(OutputFieldsDeclarer declarer){
45. }
47. @Override
48. publicMap<String,Object> getComponentConfiguration(){
49. returnnull;
50. }
51. }

提交到 Topology

• Topology概念参考《Storm学习笔记》
• KafkaStormSample.java

1. import org.apache.storm.Config;
2. import org.apache.storm.LocalCluster;
3. import org.apache.storm.topology.TopologyBuilder;
5. import java.util.ArrayList;
6. import java.util.List;
7. import java.util.UUID;
9. import org.apache.storm.spout.SchemeAsMultiScheme;
10. import org.apache.storm.kafka.trident.GlobalPartitionInformation;
11. import org.apache.storm.kafka.ZkHosts;
12. import org.apache.storm.kafka.Broker;
13. import org.apache.storm.kafka.StaticHosts;
14. import org.apache.storm.kafka.BrokerHosts;
15. import org.apache.storm.kafka.SpoutConfig;
16. import org.apache.storm.kafka.KafkaConfig;
17. import org.apache.storm.kafka.KafkaSpout;
18. import org.apache.storm.kafka.StringScheme;
20. publicclassKafkaStormSample{
21. publicstaticvoid main(String[] args)throwsException{
22. Config config =newConfig();
23. config.setDebug(true);
24. config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING,1);
25. String zkConnString ="localhost:2181";
26. String topic ="my-first-topic";
27. BrokerHosts hosts =newZkHosts(zkConnString);
29. SpoutConfig kafkaSpoutConfig =newSpoutConfig(hosts, topic,"/"+ topic, UUID.randomUUID().toString());
30. kafkaSpoutConfig.bufferSizeBytes =1024*1024*4;
31. kafkaSpoutConfig.fetchSizeBytes =1024*1024*4;
32. // kafkaSpoutConfig.forceFromStart = true;
33. kafkaSpoutConfig.scheme =newSchemeAsMultiScheme(newStringScheme());
35. TopologyBuilder builder =newTopologyBuilder();
36. builder.setSpout("kafka-spout",newKafkaSpout(kafkaSpoutConfig));
37. builder.setBolt("word-spitter",newSplitBolt()).shuffleGrouping("kafka-spout");
38. builder.setBolt("word-counter",newCountBolt()).shuffleGrouping("word-spitter");
40. LocalCluster cluster =newLocalCluster();
41. cluster.submitTopology("KafkaStormSample", config, builder.createTopology());
43. Thread.sleep(10000);
45. cluster.shutdown();
46. }
47. }

版本

• Zookeeper: zookeeper-3.5.2-alpha.tar.gz
• Curator: 2.9.1
• SLF4j: slf4j-1.7.21.tar.gz
• Kafka: kafka_2.11-0.10.1.0
• Storm : apache-storm-1.0.2.tar.gz
• JSON: json-simple-1.1.1.jar
• JDK: 1.8.0

编译

• 依赖包

  • Curator 
    Before moving compilation, Kakfa-Storm integration needs curator ZooKeeper client java library. 这个包的路径，要加到CLASSPATH中 
    
    curator-client-2.9.1.jar 
    curator-framework-2.9.1.jar 
    
  • JSON 
    json-simple-1.1.1.jar
  • Storm-Kafka 
    json-simple-1.1.1.jar
  • Kafka Lib
  • Storm Lib
  • SLF4J 
    当CLASSPATH中，包含了Kafka 和Storm 的lib 后， SLF4j会产生冲突，将Kafka中的SLF4j 移除CLASSPATH即可

• 编译命令

  1. $ javac *.java

执行

• 开启服务： Zookeeper Kafka Storm

1. $cd /opt/zookeeper
2. $.bin/zkServer.sh start
4. $cd /opt/kafka
5. $./bin/kafaka-server-start.sh config/server.properties
7. $ cd /opt/storm
8. $./bin/storm nimbus
9. $./bin/storm supervisor

• 创建Topic: “my-first-topic”

1. $ ./bin/kafktopics.sh --create --zookeeper localhost:2181--replication-factor 1--partitions 1--topic my-first-topic
2. $ ./bin/kafktopics.sh --list --zookeeper localhost:2181

• 在Kafka Producer CLI 输入message

1. $ ./bin/kafka-console-producer.sh --broker-list localhost:9092--topic my-first-topic
2. hello
3. kafka
4. storm
5. spark
6. test message
7. anther test message

• 执行例子

1. $ java KafkaStormSample

• 输出结果

1. storm :1
2. test :2
3. spark :1
4. another :1
5. kafka :1
6. hello :1
7. message :2