Распределенные вычисления в Elixir — классический пример MapReduce

Добавим в функцию задержку на 1 секунду, а также выведем результат подсчета в консоль, перед тем как его вернуть.

Тема действительно интересная, я сам недавно сталкивался с распараллеливанием вычислений в Elixir, поэтому сделал небольшой рефакторинг и добавил основную основную фичу, значительно облегчающую распараллеливание — Task.

Читать по ссылке, а пока предлагаю взглянуть на код:

defmodule Wordcount do
  @main_word "лошадь"

  defp count(text, word) do
    length(String.split(text, word)) - 1
  end

  defp do_proceed(file_path) do
    File.read!(file_path)
    |> count(@main_word)
    |> (fn (count) -> IO.puts "Found #{count} occurrence(s) of the word in file #{file_path}" end).()
  end

  def proceed(:async) do
    Path.wildcard("./OGENRI/*.txt")
    |> Enum.map(&Task.async(fn -> do_proceed(&1) end))
    |> Enum.map(&Task.await/1)
  end

  def proceed(:sync) do
    Path.wildcard("./OGENRI/*.txt")
    |> Enum.map(&do_proceed/1)
  end

  ### Example:
  #   benchmark
  #   benchmark(:sync)
  def benchmark(type \\ :async) do
    start_time = :os.system_time(:milli_seconds)
    proceed(type)
    end_time = :os.system_time(:milli_seconds)
    IO.puts "Finished in #{(end_time - start_time)} miliseconds"
  end
end

Вот конкретные различия между синхронной и асинхронной версией

def proceed(:sync) do
  Path.wildcard("./OGENRI/*.txt")
  |> Enum.map(&do_proceed/1)
end

def proceed(:async) do
  Path.wildcard("./OGENRI/*.txt")
  |> Enum.map(&Task.async(fn -> do_proceed(&1) end))
  |> Enum.map(&Task.await/1)
end

Пишем синхронный код — потом вуаля! — и он щелчком пальцев превращается в асинхронный. Добро пожаловать в functional programming!)

Path.wildcard("/data/OGENRI/*.txt")
|> Enum.map(fn(file) -> async_word_count.(file, "лошадь") end)

Ваш результат лишь показывает, что спать асинхронно в 12 потоков быстрее, чем спать 12 раз синхронно в одном.

Игрушечный пример на одной машине не показывает оверхеда от передачи данных по сети (который может быть очень значительным), управления задачами (что если нода упадёт?) и т.п.

Кроме того, "классический" MapReduce устроен посложнее, чем у вас написано: редьюсеров обычно больше одного и результаты разделяются между редьюсерами по хэшу ключа и сортируются, это как раз одна из ключевых идей. Такой подход позволяет распараллелить фазу редьюса, избежать перегрузки машины, которая выполняет редьюс (все ключи могут просто не помещаться в память), и терять меньше работы в случае аварийного завершения редьюсера.

Перемножить большие матрицы на небольшом кластере или сделать JOIN независимых датасетов по какому-нибудь общему ключу было бы гораздо более показательно.