flow-mapreduce-main.ex

def start_flow do
  streams()
  |> Flow.from_enumerables()
  |> Flow.filter(&Regex.match?(~r/publishing/, &1))
  |> Flow.map(&get_key_and_payload(&1))
  |> Flow.partition(key: {:elem, 0})
  |> Flow.reduce(fn -> %{} end, fn x, acc ->
    Map.put_new(acc, elem(x, 1), elem(x, 2))
  end)
end