# \#flow **URL:** https://forum.elixirforum.com/tag/flow/207.md [Latest](https://forum.elixirforum.com/latest.md) · [Categories](https://forum.elixirforum.com/categories.md) · [Tags](https://forum.elixirforum.com/tags.md) --- ## [Could Graph-Based Pipelines Become an Elixir Killer Feature?](https://forum.elixirforum.com/t/could-graph-based-pipelines-become-an-elixir-killer-feature/76997)
**Author:** [@brecabral](https://forum.elixirforum.com/u/brecabral)\ **Replies:** 11\ **Last updated:** [October 13, 2026, 5:12pm UTC](https://forum.elixirforum.com/t/could-graph-based-pipelines-become-an-elixir-killer-feature/76997 "2026-10-13T17:12:54Z")
From Pipes to Graphs: Could Elixir Make Control Flow More Composable? Recently, while studying LangGraph, and particularly the evolution from LangChain’s linear chains toward graph-based workflows, I started wondering: w… --- ## [Is it wise to default to always using Flow when using Broadway?](https://forum.elixirforum.com/t/is-it-wise-to-default-to-always-using-flow-when-using-broadway/69303)
**Author:** [@slouchpie](https://forum.elixirforum.com/u/slouchpie)\ **Replies:** 4\ **Last updated:** [February 14, 2025, 10:58am UTC](https://forum.elixirforum.com/t/is-it-wise-to-default-to-always-using-flow-when-using-broadway/69303 "2025-02-14T10:58:41Z")
Greetings, comrades. I have thoroughly gone through both the RabbitMQ Elixir tutorials (RabbitMQ tutorial - "Hello world!" | RabbitMQ) and the book “Concurrent Data Processing in Elixir”. GenStage is very cool, Flow is … --- ## [Elixir Flow and Windows 10](https://forum.elixirforum.com/t/elixir-flow-and-windows-10/61141)
**Author:** [@jwsinner](https://forum.elixirforum.com/u/jwsinner)\ **Replies:** 0\ **Last updated:** [January 21, 2024, 1:43am UTC](https://forum.elixirforum.com/t/elixir-flow-and-windows-10/61141 "2024-01-21T01:43:43Z")
I have a Flow setup that does a port scan on localhost to catch anything I currently have running: range |\> Flow.from\_enumerable() |\> Flow.partition() |\> Flow.map(fn port -\> %URI{ host: hos… --- ## [Data processing in Parallel (Hackathon project)](https://forum.elixirforum.com/t/data-processing-in-parallel-hackathon-project/60424)
**Author:** [@Iex.new](https://forum.elixirforum.com/u/Iex.new)\ **Replies:** 47\ **Last updated:** [January 8, 2024, 2:15pm UTC](https://forum.elixirforum.com/t/data-processing-in-parallel-hackathon-project/60424 "2024-01-08T14:15:08Z")
Hi, in our Team at work we have Hackathons and for next year I am thinking about to introduce Elixir. As we have todo with data processing in our daily work I planning to propose a small app that should import data in p… --- ## [First time trying Flow, some novice questions](https://forum.elixirforum.com/t/first-time-trying-flow-some-novice-questions/57106)
**Author:** [@ca1989](https://forum.elixirforum.com/u/ca1989)\ **Replies:** 8\ **Last updated:** [October 24, 2023, 5:15pm UTC](https://forum.elixirforum.com/t/first-time-trying-flow-some-novice-questions/57106 "2023-10-24T17:15:56Z")
Today I started studying Flow, and I have some questions. Please keep in mind that this is an example code, I want to use Flow. So, it’s: Reading a text file Remove empty lines and newlines Count the words Generate a… --- ## [Process events that depends on each other - Flow](https://forum.elixirforum.com/t/process-events-that-depends-on-each-other-flow/58691)
**Author:** [@ssss](https://forum.elixirforum.com/u/ssss)\ **Replies:** 1\ **Last updated:** [October 7, 2023, 10:34pm UTC](https://forum.elixirforum.com/t/process-events-that-depends-on-each-other-flow/58691 "2023-10-07T22:34:45Z")
Hello, I successfully implemented the whole GenStage & Flow thingy, it is really awesome, so I want to continue with it for the next job: Imagine you have a variable number of events (let’s say in chunks of 30s). They n… --- ## [Understanding GenStage/Flow Error Propagation](https://forum.elixirforum.com/t/understanding-genstage-flow-error-propagation/55652)
**Author:** [@jnnks](https://forum.elixirforum.com/u/jnnks)\ **Replies:** 4\ **Last updated:** [June 10, 2023, 5:25pm UTC](https://forum.elixirforum.com/t/understanding-genstage-flow-error-propagation/55652 "2023-06-10T17:25:35Z")
I am building a data processing pipeline which may sporadically fail. I would like to know when Producers/Consumers fail, so I can keep track of running pipelines. When I crash my producer, other processes crash with the… --- ## [Library Idea for Data Pipeline Processing](https://forum.elixirforum.com/t/library-idea-for-data-pipeline-processing/55660)
**Author:** [@Billzabob](https://forum.elixirforum.com/u/Billzabob)\ **Replies:** 3\ **Last updated:** [May 25, 2023, 6:11pm UTC](https://forum.elixirforum.com/t/library-idea-for-data-pipeline-processing/55660 "2023-05-25T18:11:28Z")
Hello! I have an idea for an Elixir library I’d like to work on, but wanted to get some thoughts from the community first. It would allow you to define data pipelines declaratively. Here’s an example of what I’m imagini… --- ## [NimbleCSV error trying to parse a large CSV partitioned in many files](https://forum.elixirforum.com/t/nimblecsv-error-trying-to-parse-a-large-csv-partitioned-in-many-files/55564)
**Author:** [@gastonrey](https://forum.elixirforum.com/u/gastonrey)\ **Replies:** 6\ **Last updated:** [April 28, 2023, 11:52am UTC](https://forum.elixirforum.com/t/nimblecsv-error-trying-to-parse-a-large-csv-partitioned-in-many-files/55564 "2023-04-28T11:52:59Z")
Hi there! I have large CSV file (2GB) so I decided to partitioned it in many files and then I’m trying to use Streams, Flow and NimbleCSV in order to parse and filter it but I’m getting an error that I can’t figure out w… --- ## [\`Stream.chunk\_every/4\` equivalent for Flow](https://forum.elixirforum.com/t/stream-chunk-every-4-equivalent-for-flow/55227)
**Author:** [@ImNotAVirus](https://forum.elixirforum.com/u/ImNotAVirus)\ **Replies:** 0\ **Last updated:** [April 12, 2023, 3:17pm UTC](https://forum.elixirforum.com/t/stream-chunk-every-4-equivalent-for-flow/55227 "2023-04-12T15:17:32Z")
Context: While practicing optimizing some exercises, I got stuck this morning trying to use Flow on a piece of code using Stream. Here is the code : counter = fn x, acc -\> Map.update(acc, x, 1, &(&1 + 1)) end # Cre… --- ## [Best way to read data in the most performant way from multiple large CSV files?](https://forum.elixirforum.com/t/best-way-to-read-data-in-the-most-performant-way-from-multiple-large-csv-files/54432)
**Author:** [@raza\_ep](https://forum.elixirforum.com/u/raza_ep)\ **Replies:** 6\ **Last updated:** [March 14, 2023, 7:43am UTC](https://forum.elixirforum.com/t/best-way-to-read-data-in-the-most-performant-way-from-multiple-large-csv-files/54432 "2023-03-14T07:43:32Z")
My objective is to read data (in the most performant approach) from multiple large CSV files, make minimal transformations and eventually write to output files. I want to ensure I am leveraging all the compute on my mach… --- ## [Does Flow.departition/5 add a new layer of stages? How to emit after departitioning?](https://forum.elixirforum.com/t/does-flow-departition-5-add-a-new-layer-of-stages-how-to-emit-after-departitioning/53091)
**Author:** [@linusdm](https://forum.elixirforum.com/u/linusdm)\ **Replies:** 0\ **Last updated:** [January 10, 2023, 5:03pm UTC](https://forum.elixirforum.com/t/does-flow-departition-5-add-a-new-layer-of-stages-how-to-emit-after-departitioning/53091 "2023-01-10T17:03:33Z")
Hi! I’m working with Flow, but I’m having trouble understanding how departitioning works. I have no mental model about how this last step really works. I understand that everything is forced back into one process. But i… --- ## [Supervising a flow built with into\_specs and through\_specs](https://forum.elixirforum.com/t/supervising-a-flow-built-with-into-specs-and-through-specs/15285)
**Author:** [@lapinkoira](https://forum.elixirforum.com/u/lapinkoira)\ **Replies:** 12\ **Last updated:** [December 20, 2022, 2:42pm UTC](https://forum.elixirforum.com/t/supervising-a-flow-built-with-into-specs-and-through-specs/15285 "2022-12-20T14:42:09Z")
With the new functions through\_specs and into\_specs you can quickly build a Flow and connect it to GenStage producer\_consumers and consumers. Something like: def start\_link(\_params) do flow |\> Flow.through\_specs(pr… --- ## [Ingesting stream of MQTT events with Flow. Is it a good idea?](https://forum.elixirforum.com/t/ingesting-stream-of-mqtt-events-with-flow-is-it-a-good-idea/51073)
**Author:** [@linusdm](https://forum.elixirforum.com/u/linusdm)\ **Replies:** 18\ **Last updated:** [October 18, 2022, 12:48pm UTC](https://forum.elixirforum.com/t/ingesting-stream-of-mqtt-events-with-flow-is-it-a-good-idea/51073 "2022-10-18T12:48:21Z")
Hey folks, I’m trying to figure out how to ingest a stream of messages from an MQTT broker in an efficient way. The problem involves many different sensors that publish a measurement once every minute. A server (one nod… --- ## [Flow and partitions](https://forum.elixirforum.com/t/flow-and-partitions/3390)
**Author:** [@leighshepperson](https://forum.elixirforum.com/u/leighshepperson)\ **Replies:** 4\ **Last updated:** [June 20, 2022, 6:23pm UTC](https://forum.elixirforum.com/t/flow-and-partitions/3390 "2022-06-20T18:23:23Z")
This is probably a basic question about flows, but: At the partition stage, where you can specify the number of partitions, does this number only relate to physical processors and nodes? Or, if the number of partitions… --- ## [Flow - Parallel hard disk read](https://forum.elixirforum.com/t/flow-parallel-hard-disk-read/47772)
**Author:** [@nfplay](https://forum.elixirforum.com/u/nfplay)\ **Replies:** 0\ **Last updated:** [May 16, 2022, 9:35am UTC](https://forum.elixirforum.com/t/flow-parallel-hard-disk-read/47772 "2022-05-16T09:35:27Z")
Hi everyone! Reading from Flow documentation around the examples that deal with processing data from streams, it’s not entirely clear to me how data is actually read from the source using Stream. From my understanding, r… --- ## [Flow + Jaxon - json stream parsing with Flow](https://forum.elixirforum.com/t/flow-jaxon-json-stream-parsing-with-flow/46551)
**Author:** [@MeerKatDev](https://forum.elixirforum.com/u/MeerKatDev)\ **Replies:** 2\ **Last updated:** [March 12, 2022, 8:53pm UTC](https://forum.elixirforum.com/t/flow-jaxon-json-stream-parsing-with-flow/46551 "2022-03-12T20:53:01Z")
Hello. So, I’m trying to stream a json file, and process it with flow: |\> File.stream! |\> Jaxon.Stream.from\_enumerable |\> Jaxon.Stream.query(\[:root, "rows", :all\]) |\> Flow.from\_enumerable() |\> Flow.p… --- ## [Flow: Implement Overlapping windows/partitions](https://forum.elixirforum.com/t/flow-implement-overlapping-windows-partitions/6007)
**Author:** [@leifg](https://forum.elixirforum.com/u/leifg)\ **Replies:** 2\ **Last updated:** [December 28, 2021, 7:00pm UTC](https://forum.elixirforum.com/t/flow-implement-overlapping-windows-partitions/6007 "2021-12-28T19:00:34Z")
Is there a way to process elements in flow in a way that a certain number if elements overlap? For example, if I have this file: 1 2 3 4 5 6 7 8 9 10 I want to process the element in this way: \[1,2\], \[2,3\], \[3, 4\], \[… --- ## [Flow based programming in elixir](https://forum.elixirforum.com/t/flow-based-programming-in-elixir/17268)
**Author:** [@sarat1669](https://forum.elixirforum.com/u/sarat1669)\ **Replies:** 23\ **Last updated:** [December 18, 2021, 9:40am UTC](https://forum.elixirforum.com/t/flow-based-programming-in-elixir/17268 "2021-12-18T09:40:16Z")
I want to know if we can implement flow based programming using a graph. Where each node is a process by itself, and the graph can be dynamic in nature. I’m not sure how I can achieve this in elixir. Please let me know y… --- ## [Flow Executing Twice](https://forum.elixirforum.com/t/flow-executing-twice/17328)
**Author:** [@gheidorn-at-cars](https://forum.elixirforum.com/u/gheidorn-at-cars)\ **Replies:** 1\ **Last updated:** [October 23, 2021, 4:43pm UTC](https://forum.elixirforum.com/t/flow-executing-twice/17328 "2021-10-23T16:43:41Z")
Flow question. I’m downloading a file from S3 where each line in the file is a JSON object. I am streaming this into a flow where I decode the JSON object to a struct, migrate (insert) to my database and produce a migr… --- ## [Why doesn't Flow have a chunk\_by function?](https://forum.elixirforum.com/t/why-doesnt-flow-have-a-chunk-by-function/43159)
**Author:** [@trisolaran](https://forum.elixirforum.com/u/trisolaran)\ **Replies:** 0\ **Last updated:** [October 18, 2021, 1:04pm UTC](https://forum.elixirforum.com/t/why-doesnt-flow-have-a-chunk-by-function/43159 "2021-10-18T13:04:38Z")
I just started using Flow and I noticed that it doesn’t have a chunk\_by function. I wonder why. Specifically, I’m reading a file containing a list of line items from multiple invoices, each line item containing an invoi… --- ## [Restricted concurrent processing options from Stream.resource/3?](https://forum.elixirforum.com/t/restricted-concurrent-processing-options-from-stream-resource-3/42651)
**Author:** [@tmock12](https://forum.elixirforum.com/u/tmock12)\ **Replies:** 1\ **Last updated:** [September 27, 2021, 5:41pm UTC](https://forum.elixirforum.com/t/restricted-concurrent-processing-options-from-stream-resource-3/42651 "2021-09-27T17:41:32Z")
I’m reading in data in batches from Ecto by using Stream.resource/3. I then map over each row and call out to a function. I can do some async, but it has to be limited as it does DB writes, etc. Let’s say 3 concurrently … --- ## [Cloning a stream: One stream, two consumers](https://forum.elixirforum.com/t/cloning-a-stream-one-stream-two-consumers/39947)
**Author:** [@hauptbenutzer](https://forum.elixirforum.com/u/hauptbenutzer)\ **Replies:** 2\ **Last updated:** [May 24, 2021, 9:06pm UTC](https://forum.elixirforum.com/t/cloning-a-stream-one-stream-two-consumers/39947 "2021-05-24T21:06:37Z")
Last week I was busy rebuilding a part of our application using streams. The pipeline is a straight-forward map-reduce with the map step involving a lot of IO-heavy work and the reduce just being aggregations on the resu… --- ## [Flow shutdown once GenStage complete](https://forum.elixirforum.com/t/flow-shutdown-once-genstage-complete/39962)
**Author:** [@peoj](https://forum.elixirforum.com/u/peoj)\ **Replies:** 0\ **Last updated:** [May 24, 2021, 8:04pm UTC](https://forum.elixirforum.com/t/flow-shutdown-once-genstage-complete/39962 "2021-05-24T20:04:23Z")
I’m writing a Flow that receives processes each item from a database cursor (Mongo in my case). Once that cursor is exhausted I’d like the flow to shut down automatically. Here’s the code I have so far: defmodule Curs… --- ## [Distributed Flows](https://forum.elixirforum.com/t/distributed-flows/39450)
**Author:** [@brent](https://forum.elixirforum.com/u/brent)\ **Replies:** 2\ **Last updated:** [May 7, 2021, 12:24pm UTC](https://forum.elixirforum.com/t/distributed-flows/39450 "2021-05-07T12:24:50Z")
Hi all, From watching GenStage and Flow - José Valim (Lambda Days 2017) at 46:00 José discussed what the next steps for Distributed support would be. I haven’t been able to find much info on how this has progressed sin… --- ## [Measuring & Tuning GenStage/Flow performance](https://forum.elixirforum.com/t/measuring-tuning-genstage-flow-performance/31078)
**Author:** [@mgwidmann](https://forum.elixirforum.com/u/mgwidmann)\ **Replies:** 1\ **Last updated:** [March 11, 2021, 8:39pm UTC](https://forum.elixirforum.com/t/measuring-tuning-genstage-flow-performance/31078 "2021-03-11T20:39:55Z")
I have personally never put a flow application into production so I was wondering what others were doing with regard to measuring and tuning GenStage and Flow specific metrics. I’ve never seen anything specifically targe… --- ## [Linking "use Flow" modules?](https://forum.elixirforum.com/t/linking-use-flow-modules/35615)
**Author:** [@cheerfulstoic](https://forum.elixirforum.com/u/cheerfulstoic)\ **Replies:** 1\ **Last updated:** [December 5, 2020, 2:59pm UTC](https://forum.elixirforum.com/t/linking-use-flow-modules/35615 "2020-12-05T14:59:26Z")
I’ve been reading documentation over and over again and trying various things, and I’m still not even sure if this is possible. A basic sanity check on this would at least be helpful :grin: I’m looking to process a str… --- ## [How to create a Flow from a non-materialized source?](https://forum.elixirforum.com/t/how-to-create-a-flow-from-a-non-materialized-source/19048)
**Author:** [@sunaku](https://forum.elixirforum.com/u/sunaku)\ **Replies:** 2\ **Last updated:** [November 18, 2020, 12:06pm UTC](https://forum.elixirforum.com/t/how-to-create-a-flow-from-a-non-materialized-source/19048 "2020-11-18T12:06:18Z")
Hello, I want to connect my flow to a few different downstream flows. However, I’m unable to instantiate the downstream flows because all Flow constructors require that you have knowledge of the upstream source (which … --- ## [Flow into Stream](https://forum.elixirforum.com/t/flow-into-stream/34527)
**Author:** [@bencoppock](https://forum.elixirforum.com/u/bencoppock)\ **Replies:** 0\ **Last updated:** [September 22, 2020, 4:31pm UTC](https://forum.elixirforum.com/t/flow-into-stream/34527 "2020-09-22T16:31:32Z")
This is my first time using Flow, and I want to make sure I’m understanding the paradigm correctly… My understanding is that the following code will Divvy out the work of mapping over the CSV lines to be done in paral… --- ## [Stream file text and parse each line information where lines are not always independent](https://forum.elixirforum.com/t/stream-file-text-and-parse-each-line-information-where-lines-are-not-always-independent/34143)
**Author:** [@raigons](https://forum.elixirforum.com/u/raigons)\ **Replies:** 2\ **Last updated:** [September 3, 2020, 4:18pm UTC](https://forum.elixirforum.com/t/stream-file-text-and-parse-each-line-information-where-lines-are-not-always-independent/34143 "2020-09-03T16:18:29Z")
Hi there. For a matter of studying and practice reasons I have the following problem: Given I have a log txt file (usually large, so that I’m using File.stream), I need to parse some lines until a pattern is found. Ex… [Next page](https://forum.elixirforum.com/tag/flow/207.md?match_all_tags=true&page=1&tags%5B%5D=flow)