Repository: storm Updated Branches: refs/heads/master dcee1e29c -> 549b01a03
When using a hash key as an option for RedisMapState, only get the values for keys in the requested batch. This commit fixes a bug whereby if the state updater is constructed with a hash key (ie, the state will be stored as a key in a redis hash, versus as a key in the top-level redis space), each call to multiGet would request the entire hash and iterate to extract only the values in the hash relevant to the batch. This can cause an inordinate amount of network traffic (and actually caused our interfaces to fall over) for states with either a moderately high cardinality or large values. Instead, the call to Redis should be an hmget (hash multiget) that takes the hash key as its first argument and an array of strings as the keys to fetch from that key, thereby retrieving only the requested values. The change also deprecates and removes buildValuesFromMap. Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/26bab159 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/26bab159 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/26bab159 Branch: refs/heads/master Commit: 26bab1595c56f6f40ea2392401e829f3ccb0cff0 Parents: 27a3606 Author: David Katten <[email protected]> Authored: Tue Mar 10 15:52:40 2015 -0500 Committer: David Katten <[email protected]> Committed: Tue Mar 10 15:52:40 2015 -0500 ---------------------------------------------------------------------- .../storm/redis/trident/state/RedisMapState.java | 19 +++++-------------- 1 file changed, 5 insertions(+), 14 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/26bab159/external/storm-redis/src/main/java/org/apache/storm/redis/trident/state/RedisMapState.java ---------------------------------------------------------------------- diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/trident/state/RedisMapState.java b/external/storm-redis/src/main/java/org/apache/storm/redis/trident/state/RedisMapState.java index f934cea..9726868 100644 --- a/external/storm-redis/src/main/java/org/apache/storm/redis/trident/state/RedisMapState.java +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/trident/state/RedisMapState.java @@ -223,8 +223,10 @@ public class RedisMapState<T> implements IBackingMap<T> { if (keys.size() == 0) { return Collections.emptyList(); } + + String[] stringKeys = buildKeys(keys); + if (Strings.isNullOrEmpty(this.options.hkey)) { - String[] stringKeys = buildKeys(keys); Jedis jedis = null; try { jedis = jedisPool.getResource(); @@ -239,8 +241,7 @@ public class RedisMapState<T> implements IBackingMap<T> { Jedis jedis = null; try { jedis = jedisPool.getResource(); - Map<String, String> keyValue = jedis.hgetAll(this.options.hkey); - List<String> values = buildValuesFromMap(keys, keyValue); + List<String> values = jedis.hmget(this.options.hkey, stringKeys); return deserializeValues(keys, values); } finally { if (jedis != null) { @@ -250,16 +251,6 @@ public class RedisMapState<T> implements IBackingMap<T> { } } - private List<String> buildValuesFromMap(List<List<Object>> keys, Map<String, String> keyValue) { - List<String> values = new ArrayList<String>(keys.size()); - for (List<Object> key : keys) { - String strKey = keyFactory.build(key); - String value = keyValue.get(strKey); - values.add(value); - } - return values; - } - private List<T> deserializeValues(List<List<Object>> keys, List<String> values) { List<T> result = new ArrayList<T>(keys.size()); for (String value : values) { @@ -303,7 +294,7 @@ public class RedisMapState<T> implements IBackingMap<T> { for (int i = 0; i < keys.size(); i++) { String val = new String(serializer.serialize(vals.get(i))); keyValues.put(keyFactory.build(keys.get(i)), val); - } + } jedis.hmset(this.options.hkey, keyValues); } finally {
