Combining the streamz.Stream.filenames() and streamz.Stream.from_textfile() using dask scatter?

Aperta
#480 2 commenti 0 reazioni 0 assegnatari Vedi su GitHub

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
5/5
Tempo stimato
Più di una settimana
Idoneità per principianti
25/100
Tipo di issue
Funzionalità
Chiarezza
Da chiarire
Stato di attività
Ferma
Stack tecnologico
python
Ambito
stream-processing

Direzione di ricerca

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.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

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()
Lingua principale
Python
Stelle
1.3k
Fork
149
Nessuna PR unita negli ultimi 30g

Guida per i contributori

Apri la guida per i contributori

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di python-streamz/streamz

Tutte le issue di python-streamz/streamz

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.