Hacktoberfest 2026: the issues maintainers tagged for October, open and beginner-friendly. Browse Hacktoberfest issues

Flatten doesn't work with DaskStream

Open
#213 5 comments 0 reactions 0 assignees View on GitHub

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

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

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

More from python-streamz/streamz

All issues in python-streamz/streamz

Similar issues

More Python issues

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.