在Clojure中并行处理CSV文件

Question

我有一个大型CSV文件,其中包含需要花费大量精力才能处理的独立项目.我希望能够并行处理每个订单项.我在这里找到了一个用于在SO上处理CSV文件的示例代码:

新手在Clojure中转换CSV文件

代码是:

(use '(clojure.contrib duck-streams str-utils))                 ;;'
(with-out-writer "coords.txt"
  (doseq [line (read-lines "coords.csv")]
    (let [[x y z p] (re-split #"," line)]
      (println (str-join \space [p x y z])))))

这能够从我的CSV文件打印出来的数据非常棒 - 但它只使用了一个CPU.我尝试了各种不同的东西,结果是:

(pmap println (read-lines "foo"))

这在交互模式下工作正常,但在从命令行运行时不执行任何操作.从IRC上的对话来看,这是因为默认情况下stdout不可用于线程.

我真正想要的是一种将函数应用于CSV文件的每一行并且并行执行此操作的方法.如果可能的话,我还想在测试期间将一些结果打印到stdout.

有任何想法吗？

Answer 1

如果您希望输出中的结果与输入中的结果顺序相同,那么从pmap打印可能不是一个好主意.我建议在其上创建输入行pmap的(延迟)序列,然后打印pmap的结果.这样的事情应该有效:

(dorun (map println (pmap expensive-computation (read-lines "coords.csv"))))

Answer 2

如果你想快速做到这一点,你可能想看看这篇文章,关于Alex Osborne如何解决Tim Bray提出的Widefinder 2挑战.Alex涉及解析,处理和收集结果的所有方面(在Widefinder 2中,该文件是一个非常大的Apache日志).使用的实际代码在这里.