并行处理多个文件

Question

我试图使用scalaz-stream同时处理多个文件，对所有文件中的每一行应用一个函数。具体而言，假设我有一个函数，该函数接受字符串列表。

def f(lines: Seq[String]): Something = ???

还有几个文件，第一个：

foo1
foo2
foo3

第二部分：

bar1
bar2
bar3

整个过程的结果应该是：

List(
  f(Seq("foo1", "bar1")), 
  f(Seq("foo2", "bar2")), 
  f(Seq("foo3", "bar3"))
)

(或者更有可能直接写入其他文件)

文件的数量事先是未知的，不同文件之间的行数可能不同，但我可以(在运行时)填充(在运行时)带有默认值的较短文件的末尾，或者删除较长的文件。

因此，本质上，我需要一种将Seq[Process[Task, String]] (通过类似io.linesR获得的)组合成单个Process[Task, Seq[String]]的方法。

实现这一目标的最简单方法是什么？

或者，更广泛地说，我如何将Process[F, I]的Process[F, I]实例组合成一个实例Process[F, Seq[I]]

我肯定有一些标准的组合器，但我想不出.

谢谢。

Answer 1

这个组合器还不存在，但是您可以添加它。我想会是这样的：

def zipN[F[_], A](xs: Seq[Process[F,A]]): Process[F,Seq[A]] = 
  if (xs.isEmpty) Process.halt
  else xs.map(_ map (Vector(_))).reduceLeft(_.zipWith(_)(_ ++ _))

您还可以添加zipAllN，它使用一个值来填充序列(它使用zipAll，以及alignN，后者允许流在耗尽时“退出”输出进程。(因此输出序列可能会变短。)

我建议您将其实现为“平衡的”减少，而不是左或右的减少，因为这样会更有效。

请提交一个拉请求+测试，如果你最终真正实现了这一点！