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 # Create a stream for large files processing "your_filename.ext" |> File.stream!() # Normalize every words |> Stream.map(&String.downcase/1) |> Stream.map(&String.replace(&1, ~r"[^a-z0-9]", " ")) # Split with spaces and remove empty words |> Stream.flat_map(&String.split(&1, " ", trim: true)) # Get every sequences of 3 words in the text |> Stream.chunk_every(3, 1, :discard) # Join the sequences |> Stream.map(&Enum.join(&1, " ")) # Count occurences for 3 words sequences |> Enum.reduce(%{}, counter) # Just displays the 10 most used sequences |> Map.to_list() |> Enum.sort_by(&elem(&1, 1), :desc) |> Enum.take(10) |> IO.inspect() # Example of output with http://www.gutenberg.org/cache/epub/2009/pg2009.txt : # [ # {"of the same", 320}, # {"the same species", 130}, # {"conditions of life", 125}, # {"in the same", 117}, # {"of natural selection", 111}, # {"from each other", 104}, # {"species of the", 102}, # {"on the other", 89}, # {"the other hand", 81}, # {"the case of", 78} # ] In short, this piece of code retrieves all the 3-word sequences of a text and displays the 10 that appear the most.
Problem:
I can’t find the equivalent of the line Stream.chunk_every(3, 1, :discard) for Flow.
Here is my current code:
Mix.install([{:flow, "~> 1.2"}]) counter = fn x, acc -> Map.update(acc, x, 1, &(&1 + 1)) end "your_filename.ext" |> File.stream!(read_ahead: 100_000) |> Flow.from_enumerable() |> Flow.map(&String.downcase/1) |> Flow.map(&String.replace(&1, ~r"[^a-z0-9]", " ")) |> Flow.flat_map(&String.split(&1, " ", trim: true)) # |> Stream.chunk_every(3, 1, :discard) <= ???????? |> Flow.map(&Enum.join(&1, " ")) |> Flow.partition() |> Flow.reduce(&Map.new/0, counter) |> Enum.to_list() |> Enum.sort_by(&elem(&1, 1), :desc) |> Enum.take(10) |> IO.inspect()