Flatten doesn't work with DaskStream
Nobody has claimed this yet.
Assessment
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Newbie friendliness
- 35/100
- Issue type
- Bug
- Clarity
- Mostly clear
- Activity status
- Stale
- Tech stack
- python
- Domain
- distributed-systems, stream-processing
Research direction
Reproduce the provided script, then inspect streamz/core.py around the flatten update at line 1020 and streamz/dask.py around line 58, where the traceback shows a Future being passed onward. Done means the scatter, partition, map, flatten, buffer, gather, and sink pipeline no longer raises the reported TypeError.
Written by the indexing model from the issue text.
Description
This code:
from __future__ import division, print_function
from time import sleep
from streamz import Stream
from dask.distributed import Client
client = Client()
def callback(datas):
print(':',datas)
return datas
source = Stream().scatter()
stream = source.partition(5)
stream = stream.map(callback)
stream = stream.flatten()
stream.buffer(15).gather().sink(print)
for i in range(30):
source.emit(i)
returns:
Traceback (most recent call last):
File "test.py", line 21, in <module>
: (0, 1, 2, 3, 4)
source.emit(i)
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 332, in emit
sync(self.loop, _)
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 1277, in sync
six.reraise(*error[0])
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 1262, in f
result[0] = yield future
File "/home/klinger/rossum/local/lib/python2.7/site-packages/tornado/gen.py", line 1099, in run
value = future.result()
File "/home/klinger/rossum/local/lib/python2.7/site-packages/tornado/concurrent.py", line 260, in result
raise_exc_info(self._exc_info)
File "/home/klinger/rossum/local/lib/python2.7/site-packages/tornado/gen.py", line 315, in wrapper
yielded = next(result)
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 327, in _
result = yield self._emit(x)
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 298, in _emit
r = downstream.update(x, who=self)
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 734, in update
return self._emit(tuple(result))
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 298, in _emit
r = downstream.update(x, who=self)
File "/home/klinger/years/2014/projects/devel/streamz/streamz/dask.py", line 58, in update
return self._emit(result)
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 298, in _emit
r = downstream.update(x, who=self)
File "/home/klinger/years/2014/projects/devel/streamz/streamz/core.py", line 1020, in update
for item in x:
TypeError: 'Future' object is not iterable
Without dask interface (removing scatter and gather), everything works well.
- Dominant language
- Python
- Stars
- 1.3k
- Forks
- 149
- Avg merge
- 17h 39m
- Merged PRs (30d)
- 1
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
More from python-streamz/streamz
-
Difficulty 3/5 1-2 days Newbie friendliness 35/100
python-streamz/streamz#481 · 4 comments ·
-
Combining the streamz.Stream.filenames() and streamz.Stream.from_textfile() using dask scatter? Open
Difficulty 5/5 Over a week Newbie friendliness 25/100
python-streamz/streamz#480 · 2 comments ·
-
Difficulty 5/5 Over a week Newbie friendliness 20/100
python-streamz/streamz#479 · 1 comment ·
-
Difficulty 5/5 Over a week Newbie friendliness 20/100
python-streamz/streamz#478 · 6 comments ·
-
Difficulty 4/5 3-5 days Newbie friendliness 10/100
python-streamz/streamz#476 · 17 comments · 2 reactions ·
All issues in python-streamz/streamz
Similar issues
-
bug
Difficulty 2/5 1-3 hours Newbie friendliness 90/100
learningequality/ricecooker#747 ·
-
Difficulty 2/5 1-3 hours Newbie friendliness 68/100
BSData/horus-heresy-3rd-edition#3171 ·
-
enhancement
Difficulty 2/5 1-3 hours Newbie friendliness 72/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 76/100
run-llama/llama_index#23199 ·
-
Difficulty 2/5 1-3 hours Newbie friendliness 84/100
KhronosGroup/glTF-Blender-IO#2769 ·