Merge pull request #399 from wwoods/master

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

Author: martindurant

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)