cblavier

cblavier

Synchronize multiple streams

Hi there,

I’m doing some computation on loads of Postgres data and I’m in the situation where I have two streams of data (Repo.stream/2) coming from different tables that I would like to merge into a single output stream containing the computed values.

What would be the best approach?

Thanks for your help!

Marked As Solved

cblavier

cblavier

It finally worked as intended! I had hard times to produce a reliable Flow Stream from my Ecto query, but then the Flow stuff is just pure delight :blush:

Here is the join code:

  defp sum_metrics_flows(flow1, flow2) do
    Flow.bounded_join(
      :full_outer,
      flow1,
      flow2,
      &elem(&1, 0),
      &elem(&1, 0),
      fn
        {date, %{contacts: c1, views: v1}}, {date, %{contacts: c2, views: v2}} ->
          {date, %{contacts: c1 + c2, views: v1 + v2}}

        date_and_metrics, nil -> date_and_metrics
        nil, date_and_metrics -> date_and_metrics
      end
    )
  end

See this thread for the producer: Code review for an Ecto GenStage Producer

Also Liked

benwilson512

benwilson512

Author of Craft GraphQL APIs in Elixir with Absinthe

What does merging these streams mean then? Are they put end to end? Intermixed? If intermixed, according to what criterion?

cblavier

cblavier

I’m doing some maths between both stream entries.

I think I finally found the way to go using Flow library, especially with https://hexdocs.pm/flow/Flow.html#bounded_join/7

It will help me to join my streams and also to improve performance by leveraging on multiple cores. I will let you know how that finally turned out!

hauleth

hauleth

Compute them within DB if possible?

Where Next?

Popular in Questions Top

srinivasu
How to handle excepions in elixir? Suppose i have A, B, C ,D, E modules. and each module has get() function. A.get() method will call th...
New
Tee
can someone please explain to me how Enum.reduce works with maps
New
Harrisonl
We have an ECS cluster with 4 services, where each task joins a single cluster, via discovery ECS discovery service. Currently when I de...
New
Jim
As a follow up to my earlier question: I have the code compiling and running but not getting a successful login from the rest server. ...
New
nsuchy
Hi. I’ve noticed that Windows Powershell has it’s own IEX command and you cannot access Elixir’s IEX due to the conflict. This isn’t a cr...
New
Fl4m3Ph03n1x
Background Let’s assume I have a typical GenServer that receives messages as requests, does some operation in a DB and returns responses....
New
Patoshizzle
After calling mix ecto.create I get this error: 17:00:32.162 [error] GenServer #PID<0.412.0> terminating ** (Postgrex.Error) FATAL...
New
sabri
Can someone explain the settings of pool_size of Ecto in config file? and what is the recommend size? Thanks
New
wernerlaude
In AR this is so simple @articles = current_user.articles How to do in Ecto? def index(conn, _params) do current_user = conn.assig...
New
lucidguppy
I have a super simple question about elixir - how would I take a file like this foo bar baz and output a new file that enumerates th...
New

Other popular topics Top

JakeBecker
TL;DR: I’ve just released an implementation of Microsoft’s IDE-independent Language Server Protocol for Elixir. It adds language support ...
1140 51847 244
New
senggen
Erlang/OTP 25 [erts-13.2.2] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] 15:22:35.803 [error] gen_event {lager_file_backend...
New
sergio
In Ruby, I can go: User.find_by(email: "foobar@email.com").update(email: "hello@email.com") How can I do something similar in Elixir? ...
New
bsollish-terakeet
Credo is smart enough to check for (something like) this: assert length(the_list) == 0 with this response: Checking if an enum is empt...
New
polypush135
As many of you may have realized by now (sorry for all the posts here) I’ve been working on a db problem where I’m trying to aggregate a ...
New
chrismccord
As promised, the first release candidate of Phoenix 1.3.0 is out! This release focuses on code generators with improved project structure...
New
minhajuddin
I have seen a lot of code which picks the first element from a list using Enum.at(0) instead of List.first. Is there a reason why people ...
New
ashish173
I am using Ecto timestamps with postgres, I can see the timestamps() use the :naive_dateime but for my use case I wanted to store the ti...
New
baxterw3b
Hi guys, i’m new in the Elixir world, and i have to say, that i love it! i’m having some problem to understand anonymous functions with ...
New
Qqwy
Original source of discussion: This topic on the Pragmatic Programmers' Functional Web Development with Elixir, OTP, and Phoenix forum. ...
New

We're in Beta

About us Mission Statement