Combining the streamz.Stream.filenames() and streamz.Stream.from_textfile() using dask scatter?
Nobody has claimed this yet.
Assessment
- Difficulty
- 5/5
- Estimated time
- Over a week
- Newbie friendliness
- 25/100
- Issue type
- Feature
- Clarity
- Needs clarification
- Activity status
- Stale
- Tech stack
- python
- Domain
- stream-processing
Research direction
Start by reading the Stream.filenames() and Stream.from_textfile() APIs and the example using dask.distributed.LocalCluster and scatter(). Determine how filenames should feed text-file streams while preserving the existing map, partition, and computation chain. Done means the combined workflow can distribute incoming channel files and process their lines continuously.
Written by the indexing model from the issue text.
Description
I want to use the streamz.Stream.filenames() API to feed the filenames parallelly to streamz.Stream.from_textfile() and do computation on the lines of the textfile till they keep coming. But the API is made in such a way that we can use either one but not both. Below is the example of the existing code written partially for a single file. Each signal I get is of 1000 values. I get 4K signals per second. per channel. This one file is for one channel.
source = streamz.Stream.from_textfile(r"~\Channel_C.txt")
def split(line):
return float(line.strip().split()[1])
pulse = source.map(split).partition(1000).map(list)
pulse.map(invert_signal).map(find_true_cfd).map(write_true_cfd_to_file)
source.start()
If the two API could be combined. Then I would like to do the following. I would distribute the files I am getting per channel. 200 such files will be there. Distribute them to dask worker threads.
from dask.distributed import LocalCluster
cluster = LocalCluster()
client = cluster.get_client()
source = streamz.Stream.filenames("*.txt").scatter().from_textfile().map(split).partition(1000).map(list).map(invert_signal).map(find_true_cfd).map(write_true_cfd_to_file)
source.start()
- Dominant language
- Python
- Stars
- 1.3k
- Forks
- 149
- PR merge metrics
- No merged PRs in 30d
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 ·
-
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 ·
-
Difficulty 4/5 3-5 days Newbie friendliness 35/100
python-streamz/streamz#475 · 3 comments ·
All issues in python-streamz/streamz
Similar issues
-
Difficulty 2/5 1-3 hours Newbie friendliness 88/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 82/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 78/100
-
enhancement
Difficulty 2/5 1-3 hours Newbie friendliness 72/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 74/100