Elixir中频率计算的快速并发算法答案

【问题标题】：fast and concurrent algorithm of frequency calculation in elixirElixir中频率计算的快速并发算法
【发布时间】：2020-05-06 14:24:58
【问题描述】：

我有两个大列表，它们的项目长度不是恒定的。每个列表都包含数百万项。我想统计second list中first list项目的频率！

例如：

a = [[c, d], [a, b, e]]
b = [[a, d, c], [e, a, b], [a, d], [c, d, a]]

# expected result of calculate_frequency(a, b) is %{[c, d] => 2, [a, b, e] => 1} Or [{[c, d], 2}, {[a, b, e], 1}]

由于列表很大，我希望这个过程同时进行。所以我写了这个函数：

  def calculate_frequency(items, data_list) do
    items
    |> Task.async_stream(
      fn item ->
        frequency =
          data_list
          |> Enum.reduce(0, fn data_row, acc ->
            if item -- data_row == [] do
              acc + 1
            else
              acc
            end
          end)

        {item, frequency}
      end,
      ordered: false
    )
    |> Enum.reduce([], fn {:ok, merged}, merged_list -> [merged | merged_list] end)
  end

但是这个算法很慢。我应该怎么做才能加快速度？

PS：请不要考虑输入输出的类型，执行速度很重要。

【问题讨论】：

Flow 是你的朋友。

标签： algorithm erlang elixir word-count

【解决方案1】：

不确定这是否足够快，当然它不是并发的。它是O(m + n)，其中m 是items 的大小，n 是data_list 的大小。我找不到更快的并发方式，因为合并所有子流程的结果也需要时间。

data_list
|> Enum.reduce(%{}, fn(item, counts)-> 
  Map.update(counts, item, 1, &(&1 + 1)) 
end)
|> Map.take(items)

仅供参考，同时做事并不一定意味着并行做事。如果你只有一个 CPU 内核，并发实际上会减慢速度，因为一个 CPU 内核一次只能做一件事。

【讨论】：

【解决方案2】：

将一个列表放入MapSet。

浏览第二个列表，查看每个元素是否在MapSet 中。

这在列表的长度上是线性的，并且两个操作应该能够并行化。

【讨论】：

【解决方案3】：

我将从规范化您要比较的数据开始，这样一个简单的相等性检查就可以判断两个项目是否像您定义的那样“相等”。根据您的代码，我猜Enum.sort/1 可以解决问题，但如果MapSet.new/1 或返回地图的函数与您的用例匹配，它可能会比较快。

defp normalize(item) do
  Enum.sort(item)
end

def calculate_frequency(items, data_list) do
  data_list = Enum.map(data_list, &normalize/1)
  items = Enum.map(items, &normalize/1)
end

如果您要从数据列表中获取大多数频率，那么我将计算数据列表的所有频率。 Elixir 1.10 引入了 Enum.frequencies/1 和 Enum.frequencies_by/2，但如果需要，您可以使用 reduce 来做到这一点。

def calculate_frequency(items, data_list) do
  data_frequencies = Enum.frequencies_by(data_list, &normalize/1) # does map for you

  Map.new(items, &Map.get(data_frequencies, normalize(&1), 0)) # if you want result as map
end

我没有对我的代码或你的代码进行任何基准测试。如果你想做更多的异步操作，你可以用Task.async_stream/3替换你的映射，你可以用Stream.chunk_every/2、Task.async_stream/3的组合替换你的频率调用（Enum.frequencies/1是函数）和@ 987654331@.

【讨论】：