Storm 与 Redis 整合

Storm 与 Redis 整合

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

 

Storm 与 Redis 整合
参考资料：http://storm.apache.org/releases/2.0.0-SNAPSHOT/storm-redis.html
Strom为Redis提供几个标准Bolt
RedisLookupBolt 例子
RedisStoreBoltBolt 例子
非简单Bolt
RedisLookupBolt 源代码，用于理解WordCountRedisLookupMapper
Kafka + Storm + Redis 完整例子
依赖包
编译执行

 

参考资料：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);
  • 对例子理解

  • Storm 与 Redis 整合
     
    如上图，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

相关文章：

  • 2022-12-23
  • 2022-01-24
  • 2022-12-23
  • 2022-03-08
  • 2021-12-26
  • 2021-11-12
  • 2021-09-21
  • 2021-05-27
猜你喜欢
  • 2021-11-19
  • 2021-10-05
  • 2021-12-27
  • 2022-12-23
  • 2021-11-19
  • 2021-09-09
  • 2021-10-01
相关资源
相似解决方案
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode