Storm 与 Redis 整合

原创编写： 王宇 
2016-11-07

参考资料：http://storm.apache.org/releases/2.0.0-SNAPSHOT/storm-redis.html

Strom为Redis提供几个标准Bolt

• RedisLookupBolt: 查询
• RedisStoreBolt: 存储
• AbstractRedisBolt: 存储

RedisLookupBolt 例子

从Redis中，查询单词的计算数量

1. classWordCountRedisLookupMapperimplementsRedisLookupMapper{
2. privateRedisDataTypeDescription description;
3. privatefinalString hashKey ="wordCount";
5. publicWordCountRedisLookupMapper(){
6. description =newRedisDataTypeDescription(
7. RedisDataTypeDescription.RedisDataType.HASH, hashKey);
8. }
10. @Override
11. publicList<Values> toTuple(ITuple input,Object value){
12. String member = getKeyFromTuple(input);
13. List<Values> values =Lists.newArrayList();
14. values.add(newValues(member, value));
15. return values;
16. }
18. @Override
19. publicvoid declareOutputFields(OutputFieldsDeclarer declarer){
20. declarer.declare(newFields("wordName","count"));
21. }
23. @Override
24. publicRedisDataTypeDescription getDataTypeDescription(){
25. return description;
26. }
28. @Override
29. publicString getKeyFromTuple(ITuple tuple){
30. return tuple.getStringByField("word");
31. }
33. @Override
34. publicString getValueFromTuple(ITuple tuple){
35. returnnull;
36. }
37. }
39. JedisPoolConfig poolConfig =newJedisPoolConfig.Builder()
40. .setHost(host).setPort(port).build();
41. RedisLookupMapper lookupMapper =newWordCountRedisLookupMapper();
42. RedisLookupBolt lookupBolt =newRedisLookupBolt(poolConfig, lookupMapper);

• 对例子理解
• 
  
   
  如上图，JedisPoolConfig的作用是，提供Redis相关的配置，给RedisLookupBolt 
  RedisLookupMapper的作用是，绘制相应的数据结构给RedisLookupBolt. 
  例如
  • getKeyFromTuple()方法， 告诉RedisLookupBolt,从输入的Tuple中，取什么样子的Key，让它去Redis中查询。
  • declareOutputFields()的作用是，定义RedisLookupBolt的输出格式。

RedisStoreBoltBolt 例子

2. classWordCountStoreMapperimplementsRedisStoreMapper{
3. privateRedisDataTypeDescription description;
4. privatefinalString hashKey ="wordCount";
6. publicWordCountStoreMapper(){
7. description =newRedisDataTypeDescription(
8. RedisDataTypeDescription.RedisDataType.HASH, hashKey);
9. }
11. @Override
12. publicRedisDataTypeDescription getDataTypeDescription(){
13. return description;
14. }
16. @Override
17. publicString getKeyFromTuple(ITuple tuple){
18. return tuple.getStringByField("word");
19. }
21. @Override
22. publicString getValueFromTuple(ITuple tuple){
23. return tuple.getStringByField("count");
24. }
25. }
27. JedisPoolConfig poolConfig =newJedisPoolConfig.Builder()
28. .setHost(host).setPort(port).build();
29. RedisStoreMapper storeMapper =newWordCountStoreMapper();
30. RedisStoreBolt storeBolt =newRedisStoreBolt(poolConfig, storeMapper);

非简单Bolt

采用 AbstractRedisBolt 实现更为复杂的逻辑

1. publicstaticclassLookupWordTotalCountBoltextendsAbstractRedisBolt{
2. privatestaticfinalLogger LOG =LoggerFactory.getLogger(LookupWordTotalCountBolt.class);
3. privatestaticfinalRandom RANDOM =newRandom();
5. publicLookupWordTotalCountBolt(JedisPoolConfig config){
6. super(config);
7. }
9. publicLookupWordTotalCountBolt(JedisClusterConfig config){
10. super(config);
11. }
13. @Override
14. publicvoid execute(Tuple input){
15. JedisCommands jedisCommands =null;
16. try{
17. jedisCommands = getInstance();
18. String wordName = input.getStringByField("word");
19. String countStr = jedisCommands.get(wordName);
20. if(countStr !=null){
21. int count =Integer.parseInt(countStr);
22. this.collector.emit(newValues(wordName, count));
24. // print lookup result with low probability
25. if(RANDOM.nextInt(1000)>995){
26. LOG.info("Lookup result - word : "+ wordName +" / count : "+ count);
27. }
28. }else{
29. // skip
30. LOG.warn("Word not found in Redis - word : "+ wordName);
31. }
32. }finally{
33. if(jedisCommands !=null){
34. returnInstance(jedisCommands);
35. }
36. this.collector.ack(input);
37. }
38. }
40. @Override
41. publicvoid declareOutputFields(OutputFieldsDeclarer declarer){
42. // wordName, count
43. declarer.declare(newFields("wordName","count"));
44. }
45. }

RedisLookupBolt 源代码，用于理解WordCountRedisLookupMapper

注意下列代码中的：

• get

1. package org.apache.storm.redis.bolt;
3. import org.apache.storm.topology.OutputFieldsDeclarer;
4. import org.apache.storm.tuple.Tuple;
5. import org.apache.storm.tuple.Values;
6. import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
7. import org.apache.storm.redis.common.mapper.RedisLookupMapper;
8. import org.apache.storm.redis.common.config.JedisClusterConfig;
9. import org.apache.storm.redis.common.config.JedisPoolConfig;
10. import redis.clients.jedis.JedisCommands;
12. import java.util.List;
14. /**
15. * Basic bolt for querying from Redis and emits response as tuple.
16. * <p/>
17. * Various data types are supported: STRING, LIST, HASH, SET, SORTED_SET, HYPER_LOG_LOG, GEO
18. */
19. publicclassRedisLookupBoltextendsAbstractRedisBolt{
20. privatefinalRedisLookupMapper lookupMapper;
21. privatefinalRedisDataTypeDescription.RedisDataType dataType;
22. privatefinalString additionalKey;
24. /**
25. * Constructor for single Redis environment (JedisPool)
26. * @param config configuration for initializing JedisPool
27. * @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses
28. */
29. publicRedisLookupBolt(JedisPoolConfig config,RedisLookupMapper lookupMapper){
30. super(config);
32. this.lookupMapper = lookupMapper;
34. RedisDataTypeDescription dataTypeDescription = lookupMapper.getDataTypeDescription();
35. this.dataType = dataTypeDescription.getDataType();
36. this.additionalKey = dataTypeDescription.getAdditionalKey();
37. }
39. /**
40. * Constructor for Redis Cluster environment (JedisCluster)
41. * @param config configuration for initializing JedisCluster
42. * @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses
43. */
44. publicRedisLookupBolt(JedisClusterConfig config,RedisLookupMapper lookupMapper){
45. super(config);
47. this.lookupMapper = lookupMapper;
49. RedisDataTypeDescription dataTypeDescription = lookupMapper.getDataTypeDescription();
50. this.dataType = dataTypeDescription.getDataType();
51. this.additionalKey = dataTypeDescription.getAdditionalKey();
52. }
54. /**
55. * {@inheritDoc}
56. */
57. @Override
58. publicvoid execute(Tuple input){
59. String key = lookupMapper.getKeyFromTuple(input);
60. Object lookupValue;
62. JedisCommands jedisCommand =null;
63. try{
64. jedisCommand = getInstance();
66. switch(dataType){
67. case STRING:
68. lookupValue = jedisCommand.get(key);
69. break;
71. case LIST:
72. lookupValue = jedisCommand.lpop(key);
73. break;
75. case HASH:
76. lookupValue = jedisCommand.hget(additionalKey, key);
77. break;
79. case SET:
80. lookupValue = jedisCommand.scard(key);
81. break;
83. case SORTED_SET:
84. lookupValue = jedisCommand.zscore(additionalKey, key);
85. break;
87. case HYPER_LOG_LOG:
88. lookupValue = jedisCommand.pfcount(key);
89. break;
91. case GEO:
92. lookupValue = jedisCommand.geopos(additionalKey, key);
93. break;
95. default:
96. thrownewIllegalArgumentException("Cannot process such data type: "+ dataType);
97. }
99. List<Values> values = lookupMapper.toTuple(input, lookupValue);
100. for(Values value : values){
101. collector.emit(input, value);
102. }
104. collector.ack(input);
105. }catch(Exception e){
106. this.collector.reportError(e);
107. this.collector.fail(input);
108. }finally{
109. returnInstance(jedisCommand);
110. }
111. }
113. /**
114. * {@inheritDoc}
115. */
116. @Override
117. publicvoid declareOutputFields(OutputFieldsDeclarer declarer){
118. lookupMapper.declareOutputFields(declarer);
119. }
120. }

Kafka + Storm + Redis 完整例子

• 场景 
  参考 《Storm与Kafka整合记录》中，统计单词数量的例子。在这个例子基础之上，将统计到单词结果保存到Redis中。
• 修改Topology，增加RedisStoreBolt

1. publicclassKafkaStormSample{
2. publicclassKafkaStormSample{
3. publicstaticvoid main(String[] args)throwsException{
4. Config config =newConfig();
5. config.setDebug(true);
6. config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING,1);
7. String zkConnString ="localhost:2181";
8. String topic ="my-first-topic";
9. BrokerHosts hosts =newZkHosts(zkConnString);
11. SpoutConfig kafkaSpoutConfig =newSpoutConfig(hosts, topic,"/"+ topic, UUID.randomUUID().toString());
12. kafkaSpoutConfig.bufferSizeBytes =1024*1024*4;
13. kafkaSpoutConfig.fetchSizeBytes =1024*1024*4;
14. // kafkaSpoutConfig.forceFromStart = true;
15. kafkaSpoutConfig.scheme =newSchemeAsMultiScheme(newStringScheme());
17. // Creating RedisStoreBolt
18. String host ="localhost";
19. int port =6379;
21. JedisPoolConfig poolConfig =newJedisPoolConfig.Builder().setHost(host).setPort(port).build();
23. RedisStoreMapper storeMapper =newWordCountStoreMapper();
24. RedisStoreBolt storeBolt =newRedisStoreBolt(poolConfig, storeMapper);
26. // Assemble with topology
27. TopologyBuilder builder =newTopologyBuilder();
28. builder.setSpout("kafka-spout",newKafkaSpout(kafkaSpoutConfig));
29. builder.setBolt("word-spitter",newSplitBolt()).shuffleGrouping("kafka-spout");
30. builder.setBolt("word-counter",newCountBolt()).shuffleGrouping("word-spitter");
31. builder.setBolt("redis-store-bolt", storeBolt).shuffleGrouping("word-counter");
33. LocalCluster cluster =newLocalCluster();
34. cluster.submitTopology("KafkaStormSample", config, builder.createTopology());
36. Thread.sleep(10000);
38. cluster.shutdown();
39. }
40. }

• 修改 CountBolt, emit to RedisStoreBolt

1. publicclassCountBoltimplementsIRichBolt{
2. Map<String,Integer> counters;
3. privateOutputCollector collector;
5. @Override
6. publicvoid prepare(Map stormConf,TopologyContext context,
7. OutputCollector collector){
8. this.counters =newHashMap<String,Integer>();
9. this.collector = collector;
10. }
12. @Override
13. publicvoid execute(Tuple input){
14. String str = input.getString(0);
16. if(!counters.containsKey(str)){
17. counters.put(str,1);
18. }else{
19. Integer c = counters.get(str)+1;
20. counters.put(str, c);
21. }
23. // emit to redis-store-bolt
24. Integer result = counters.get(str);
25. this.collector.emit(newValues(str,String.valueOf(result)));
27. collector.ack(input);
28. }
30. @Override
31. publicvoid cleanup(){
32. for(Map.Entry<String,Integer> entry:counters.entrySet()){
33. System.out.println(entry.getKey()+" : "+ entry.getValue());
34. }
35. }
37. @Override
38. publicvoid declareOutputFields(OutputFieldsDeclarer declarer){
39. declarer.declare(newFields("word","count"));
41. }
43. @Override
44. publicMap<String,Object> getComponentConfiguration(){
45. returnnull;
46. }
47. }

依赖包

jedis-2.9.0.jar 
commons-pool2-2.4.2.jar

将以上两个包，增加到CLASSPATH路径中

编译执行

• 启动服务

  • 启动 zookeeper
    1. $ cd /opt/zookeeper
    2. $ ./bin/zkServer.sh start
  • 启动 redis
    1. $ redis-server
  • 启动 kafka
    1. $ cd /opt/kafka
    2. $ ./bin/kafka-server-start.sh config/server.properties
  • 启动 storm
    1. $ cd /opt/storm
    2. $./bin/storm nimbus
    3. $./bin/storm supervisor

• 输入数据

1. $cd /opt/kafka
2. $./bin/kafka-console-producer.sh --broker-list localhost:9092--topic my-first-topic
3. Hello
4. My first message
5. My second message

• 执行例子

1. $ java KafkaStormSample

• 在Redis-cli 中查询结果

1. $ redis-cli
2. 127.0.0.1:6379> hvals wordCount