使用 Elixir 将 Enum 转换为 Flow答案

【问题标题】：Convert Enum into Flow with elixir使用 Elixir 将 Enum 转换为 Flow
【发布时间】：2019-06-12 18:25:58
【问题描述】：

我有 200 万个数据文件要处理。以下代码需要 2 小时才能完成。

out_file = "./output.tsv"
result = "./input.tsv"
         |> File.stream!
         |> CSV.decode(separator: ?\t, headers: headers)
         |> Enum.map(&(elem(&1, 1)))
         |> Enum.group_by(&{&1.id, &1.name})
         |> Enum.map(&(format_data(&1)))
File.write(out_file, result)

为了获得高性能，我采用了Flow，然后我编写了以下代码。好像是Enum，所以干脆加了Flow.from_enumerable，然后把Enum改写成Flow。

out_file = "./output.tsv"
result = "./input.tsv"
         |> File.stream!
         |> CSV.decode(separator: ?\t, headers: headers)
         |> Flow.from_enumerable(stages: 4)
         |> Flow.map(&(elem(&1, 1)))
         |> Flow.group_by(&{&1.id, &1.name})
         |> Flow.map(&(format_data(&1)))
File.write(out_file, result)

它不起作用。我不认为像这样使用Flow 的正确方法。请让我知道您对正确使用Flow 的建议。

【问题讨论】：

标签： elixir

【解决方案1】：

Flow 主文档页面上的最上面的示例显示您必须终止 Flow 以及 Stream 有点像 Enum.to_list()。

在您的第一个 sn-p 中，终止发生在第一次调用 Enum.map/2 时（Stream.map/2 和家庭将用于流处理，File.stream!/1 与 Fire.read/1 的行为相同，因为您立即终止它.)

此外，NimbleCSV 由 Elixir 核心团队明确创建，用于流式处理 CSV。无论如何，下面可能会起作用：

result =
  "./input.tsv"
  |> File.stream!
  # here the stream is terminated
  |> CSV.decode(separator: ?\t, headers: headers)
  |> Flow.from_enumerable(stages: 4)
  |> Flow.map(&(elem(&1, 1)))
  |> Flow.group_by(&{&1.id, &1.name})
  |> Flow.map(&(format_data(&1)))
  # ⇓ THIS IS IMPORTANT
  |> Enum.to_list()

更好的方法是：

result =
  "./input.tsv"
  |> File.stream!(read_ahead: 100_000)
  |> NimbleCSV.RFC4180.parse_stream()
  |> Flow.from_enumerable(stages: 4)
  |> Flow.map(&(elem(&1, 1)))
  |> Flow.group_by(&{&1.id, &1.name})
  |> Flow.map(&(format_data(&1)))
  # ⇓ THIS IS IMPORTANT
  |> Enum.to_list()

【讨论】：