diff --git a/setup.py b/setup.py
index 469b2751..7d91c931 100644
--- a/setup.py
+++ b/setup.py
@@ -23,7 +23,7 @@
 
 extras_require = {
     'test': tests_require,
-    'redis': ['redis>=2.8.0'],
+    'redis': ['redis>=3.2.0'],
     'cassandra': ['cassandra-driver>=2.7.2'],
 }
 
diff --git a/stream_framework/storage/redis/structures/base.py b/stream_framework/storage/redis/structures/base.py
index 7c942fdb..a2c01742 100644
--- a/stream_framework/storage/redis/structures/base.py
+++ b/stream_framework/storage/redis/structures/base.py
@@ -1,5 +1,5 @@
 from stream_framework.storage.redis.connection import get_redis_connection
-from redis.client import BasePipeline
+from redis.client import Pipeline
 
 
 class RedisCache(object):
@@ -49,7 +49,7 @@ def _pipeline_if_needed(self, operation, *args, **kwargs):
         If the redis connection is already in distributed state use it
         Otherwise spawn a new distributed connection using .map
         '''
-        pipe_needed = not isinstance(self.redis, BasePipeline)
+        pipe_needed = not isinstance(self.redis, Pipeline)
         if pipe_needed:
             pipe = self.redis.pipeline(transaction=False)
             operation(pipe, *args, **kwargs)
diff --git a/stream_framework/storage/redis/structures/sorted_set.py b/stream_framework/storage/redis/structures/sorted_set.py
index bb2d9e33..a776fd6e 100644
--- a/stream_framework/storage/redis/structures/sorted_set.py
+++ b/stream_framework/storage/redis/structures/sorted_set.py
@@ -63,7 +63,8 @@ def _add_many(redis, score_value_pairs):
             score_value_chunks = chunks(score_value_list, 200)
 
             for score_value_chunk in score_value_chunks:
-                result = redis.zadd(key, *score_value_chunk)
+                # redis >3.2 requires a dictionary 
+                result = redis.zadd(key, {k:v for (v, k) in [score_value_chunk]})
                 logger.debug('adding to %s with score_value_chunk %s',
                              key, score_value_chunk)
                 results.append(result)
diff --git a/stream_framework/tests/storage/redis/structures.py b/stream_framework/tests/storage/redis/structures.py
index 89db0fd2..3f9e1f4c 100644
--- a/stream_framework/tests/storage/redis/structures.py
+++ b/stream_framework/tests/storage/redis/structures.py
@@ -163,11 +163,11 @@ def test_zremrangebyrank(self):
         key = 'test'
         # start out fresh
         redis.delete(key)
-        redis.zadd(key, 1, 'a')
-        redis.zadd(key, 2, 'b')
-        redis.zadd(key, 3, 'c')
-        redis.zadd(key, 4, 'd')
-        redis.zadd(key, 5, 'e')
+        redis.zadd(key, {1, 'a'})
+        redis.zadd(key, {2, 'b'})
+        redis.zadd(key, {3, 'c'})
+        redis.zadd(key, {4, 'd'})
+        redis.zadd(key, {5, 'e'})
         expected_results = [('a', 1.0), ('b', 2.0), ('c', 3.0), (
             'd', 4.0), ('e', 5.0)]
         results = redis.zrange(key, 0, -1, withscores=True)