You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2015/02/23 17:26:25 UTC
[12/23] storm git commit: change readme for storm-redis
change readme for storm-redis
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d5267c7f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d5267c7f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d5267c7f
Branch: refs/heads/master
Commit: d5267c7f9463df818ee2e6842341617448081741
Parents: 0f9438f
Author: dashengju <da...@qq.com>
Authored: Sun Jan 4 17:06:25 2015 +0800
Committer: dashengju <da...@qq.com>
Committed: Sun Jan 4 17:06:25 2015 +0800
----------------------------------------------------------------------
external/storm-redis/README.md | 98 +++++++++++++++++++++++++++++++++++++
1 file changed, 98 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d5267c7f/external/storm-redis/README.md
----------------------------------------------------------------------
diff --git a/external/storm-redis/README.md b/external/storm-redis/README.md
index b8a1458..06f3940 100644
--- a/external/storm-redis/README.md
+++ b/external/storm-redis/README.md
@@ -17,6 +17,104 @@ use it as a maven dependency:
</dependency>
```
+### AbstractRedisBolt usage:
+
+```java
+
+ public static class LookupWordTotalCountBolt extends AbstractRedisBolt {
+ private static final Logger LOG = LoggerFactory.getLogger(LookupWordTotalCountBolt.class);
+ private static final Random RANDOM = new Random();
+
+ public LookupWordTotalCountBolt(JedisPoolConfig config) {
+ super(config);
+ }
+
+ public LookupWordTotalCountBolt(JedisClusterConfig config) {
+ super(config);
+ }
+
+ @Override
+ public void execute(Tuple input) {
+ JedisCommands jedisCommands = null;
+ try {
+ jedisCommands = getInstance();
+ String wordName = input.getStringByField("word");
+ String countStr = jedisCommands.get(wordName);
+ if (countStr != null) {
+ int count = Integer.parseInt(countStr);
+ this.collector.emit(new Values(wordName, count));
+
+ // print lookup result with low probability
+ if(RANDOM.nextInt(1000) > 995) {
+ LOG.info("Lookup result - word : " + wordName + " / count : " + count);
+ }
+ } else {
+ // skip
+ LOG.warn("Word not found in Redis - word : " + wordName);
+ }
+ } finally {
+ if (jedisCommands != null) {
+ returnInstance(jedisCommands);
+ }
+ this.collector.ack(input);
+ }
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ // wordName, count
+ declarer.declare(new Fields("wordName", "count"));
+ }
+ }
+
+```
+
+### Trident State usage
+
+1. RedisState and RedisMapState, which provide Jedis interface just for single redis.
+
+2. RedisClusterState and RedisClusterMapState, which provide JedisCluster interface, just for redis cluster.
+
+3. ShardedRedisState and ShardedRedisMapState, which provide ShardedJedis interface, but not implements now.
+
+RedisState
+```java
+ JedisPoolConfig poolConfig = new JedisPoolConfig.Builder()
+ .setHost(redisHost).setPort(redisPort)
+ .build();
+ TridentTupleMapper tupleMapper = new WordCountTupleMapper();
+ RedisState.Factory factory = new RedisState.Factory(poolConfig);
+
+ TridentTopology topology = new TridentTopology();
+ Stream stream = topology.newStream("spout1", spout);
+
+ stream.partitionPersist(factory,
+ fields,
+ new RedisStateUpdater("test_", tupleMapper, 86400000),
+ new Fields());
+```
+
+RedisClusterState
+```java
+ Set<InetSocketAddress> nodes = new HashSet<InetSocketAddress>();
+ for (String hostPort : redisHostPort.split(",")) {
+ String[] host_port = hostPort.split(":");
+ nodes.add(new InetSocketAddress(host_port[0], Integer.valueOf(host_port[1])));
+ }
+ JedisClusterConfig clusterConfig = new JedisClusterConfig.Builder().setNodes(nodes)
+ .build();
+ TridentTupleMapper tupleMapper = new WordCountTupleMapper();
+ RedisClusterState.Factory factory = new RedisClusterState.Factory(clusterConfig);
+
+ TridentTopology topology = new TridentTopology();
+ Stream stream = topology.newStream("spout1", spout);
+
+ stream.partitionPersist(factory,
+ fields,
+ new RedisClusterStateUpdater("test_", tupleMapper, 86400000),
+ new Fields());
+```
+
## License
Licensed to the Apache Software Foundation (ASF) under one