nallwhy

nallwhy

How can I get last value of stream?

I need to read very large data and write that to csv file.
I want to get last datum of the data after writing.

file = File.open!(file_path, [:write, :utf8])

Stream.unfold(start_id, fn
  from_id ->
    logs = Log.list(%{from_id: from_id, size: @size})

    last_id =
      case logs |> List.last() do
        nil -> nil
        last_log -> last_log |> Map.get(:id)
      end

    {logs, last_id}
end)
|> Stream.flat_map(fn logs ->
  logs |> to_list()
end)
|> CSV.encode()
|> some_func() <- This is what I want

1039292

using Enum.to_list() |> List.last() takes too large resource.

simple version

For example:

[1, 2, 3]
|> Stream.map(fn x -> x * 2 end)
|> some_func()

6

Most Liked

evadne

evadne

You are indeed correct as Stream.transform/4 with an agent would do the job correctly.

enum = [1, 2, 3, 4, 5]
{:ok, agent_pid} = Agent.start_link(fn -> nil end)
start_fun = fn -> nil end
reducer_fun = fn v, _ -> {[], v} end
after_fun = fn v -> Agent.update(agent_pid, fn _ -> v end) end
stream = Stream.transform(enum, start_fun, reducer_fun, after_fun)
:ok = Stream.run(stream)
value = Agent.get(agent_pid, & &1)
:ok = Agent.stop(agent_pid)
value

FYI Elixir's Stream.transform doesn't emit its accumulator at the end of the input. What to do? · GitHub

mudasobwa

mudasobwa

Creator of Cure

I doubt it’s what was requested. According to docs on Stream.take/2.

If a negative count is given, the last count values will be taken. For such, the collection is fully enumerated keeping up to 2 * count elements in memory.


I would use Stream.transform/4 instead of Stream.unfold/2.

sasajuric

sasajuric

Author of Elixir In Action

The last element of an enumerable can be fetched with Enum.at(enumerable, -1). However this won’t help you here. Conceptually the problem can be described as:

# at the input we have unencoded table data (enumerable of lists)
|> CSV.encode()
# at the output we have a stream of encoded rows

So whatever additional info you collect (like e.g. id of the last row), it has to be discarded before CSV.encode, and at the output you get a stream of encoded rows.

For this particular case I’d try something like this (untested):

unencoded_rows =
  # using Stream.resource to avoid `Stream.unfold` + `Stream.flat_map`
  Stream.resource(
    fn -> start_id end, 
    fn from_id ->
      case Log.list(%{from_id: from_id, size: @size}) do
        [] -> {:halt, nil}
        logs -> {logs, logs |> List.last() |> Map.get(:id)}
      end
    end,
    fn _ -> :ok end
  )

encoded_ids =
  # using `Stream.transform` to ensure the file is open only while the stream is being consumed
  Stream.transform(
    unencoded_rows,
    fn -> File.open!(file_path, [:write, :utf8]) end,
    fn log, file ->
      # encode to csv and store as a side-effect
      csv_row = [log] |> CSV.encode() |> Enum.to_list()
      IO.write(file, csv_row)

      # return the id of the encoded entry
      {[Map.get(log, :id)], file}
    end,
    &File.close/1
  )

# force the entire stream to run and get the last entry
last_encoded_id = Enum.at(encoded_ids, -1)

In other words, we encode each row, store it to file, and keep track of the id of the stored rows. You could replace Stream.transform + Enum.at with Enum.reduce, but then you need to open file before Enum.reduce, and use try/after to ensure the file is closed immediately after.

evadne

evadne

Use Stream.take/2 with -1 as count as long as the Stream is not infinite

mudasobwa

mudasobwa

Creator of Cure

FWIW, 4 years ago I even proposed the PR to the core to make transform/4 to preserve an accumulator, but it was ruled out in favour of chunk_×××/× family to deal with it.

That said, we might abuse Stream.chunk_while/4 to do what we want.

[1, 2, 3]
|> Stream.chunk_while(
  nil,
  fn e, _ -> {:cont, e} end, # do whatever here, return `e`
  fn acc -> {:cont, acc, acc} end
)
|> Enum.to_list()
#⇒ [3]

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
pmjoe
I have a relationship of love and hate with Elixir. Lots of things are just absolutely right, but there are some things that are kind of ...
New
stefanchrobot
What’s the safe way to decode a JSON string into a struct? I want to avoid calling String.to_atom. Jason.decode can give me a map with st...
New
myronmarston
The Elixir Typespec docs show the following syntax for keyword lists in typespecs: # ... | [key: type] # keyword lis...
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
chewm
Hi guys, nice to meet you to the whole forum, I’m new here, I’m trying to configure visual studio code for elixir, right now the intellis...
New
chrisalley
ExUnit now has describe blocks which is a welcome addition coming from RSpec. In the docs, it states that nested hierarchies of describe ...
New
belgoros
I’m not a pro in using Regex and can’t figure out why the following behaviour happens, especially if we take into account the difference ...
New
sabri
Can someone explain the settings of pool_size of Ecto in config file? and what is the recommend size? Thanks
New
siddhant3030
Hi, I have to write a raw query for one of my project. But till now I have used ecto queries and don’t have much experience writing raw ...
New

Other popular topics Top

Qqwy
Update: How to use the Blogs &amp; Podcasts section You can post links to your blog posts or podcasts either in one of the Official Blog...
3268 119930 1237
New
sorentwo
Hello! tl;dr Announcing Oban, an Ecto based job processing library with a focus on reliability and historical observability. After spen...
977 41022 311
New
joaquinalcerro
Hi there, I am working with Ecto-Postgresql and I need to call all of the records from a specific table but the table has 40,000 record...
New
KronicDeth
Elixir plugin for JetBrain’s IntelliJ Platform (including Rubymine) This is a plugin that adds support for Elixir to JetBrains IntelliJ...
289 35421 110
New
chrismccord
This release brings a number of exciting features, including integration with the new Phoenix LiveDashboard and Phoenix LiveView. There h...
New
myronmarston
The Elixir Typespec docs show the following syntax for keyword lists in typespecs: # ... | [key: type] # keyword lis...
New
chensan
I have a User schema with a :from_id field set to type :string: defmodule TweetBot.Repo.Migrations.CreateUsers do use Ecto.Migration ...
New
johnnyicon
Hi all, I've just started learning Elixir and Phoenix Framework, so please pardon my n00bness at this stage. I'm trying to use Postg...
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
belgoros
I’m not a pro in using Regex and can’t figure out why the following behaviour happens, especially if we take into account the difference ...
New

We're in Beta

About us Mission Statement