Fix for #394 #397 - non-unique values with scatter

Before this, non-unique values being emitted to scatter would result in
an underlying Dask error. This fix ensures that even duplicated values
will each be executed in the stream, an idea which is more intuitive and
obviates the need for the "pure" specification used in standard Dask.

Author: wwoods

streamz/dask.py

@@ -5,7 +5,6 @@
 from tornado import gen
 
 from dask.compatibility import apply
-from dask.base import tokenize
 from distributed.client import default_client
 
 from .core import Stream
@@ -106,12 +105,11 @@ def update(self, x, who=None, metadata=None):
         self._retain_refs(metadata)
         # We need to make sure that x is treated as it is by dask
         # However, client.scatter works internally different for
-        # lists and dicts. So we always use a dict here to be sure
-        # we know the format exactly. The key will be taken as the
-        # dask identifier of the data.
-        tokenized_x = f"{type(x).__name__}-{tokenize(x)}"
-        future_as_dict = yield client.scatter({tokenized_x: x}, asynchronous=True)
-        future = future_as_dict[tokenized_x]
+        # lists and dicts. So we always use a list here to be sure
+        # we know the format exactly. We do not use a key to avoid
+        # issues like https://github.com/python-streamz/streams/issues/397.
+        future_as_list = yield client.scatter([x], asynchronous=True, hash=False)
+        future = future_as_list[0]
         f = yield self._emit(future, metadata=metadata)
         self._release_refs(metadata)
 

streamz/tests/test_dask.py

@@ -1,4 +1,5 @@
 from operator import add
+import random
 import time
 
 import pytest
@@ -50,6 +51,24 @@ def add_to_dict(d):
         assert item["i"] == i
 
 
+@gen_cluster(client=True)
+def test_non_unique_emit(c, s, a, b):
+    """Regression for https://github.com/python-streamz/streams/issues/397
+
+    Non-unique stream entries still need to each be processed.
+    """
+    source = Stream(asynchronous=True)
+    futures = source.scatter().map(lambda x: random.random())
+    L = futures.gather().sink_to_list()
+
+    for _ in range(3):
+        # Emit non-unique values
+        yield source.emit(0)
+
+    assert len(L) == 3
+    assert L[0] != L[1] or L[0] != L[2]
+
+
 @gen_cluster(client=True)
 def test_scan(c, s, a, b):
     source = Stream(asynchronous=True)