scala - 将输入流式传输到 Scala 中的外部进程

Question

我有一个 Iterable[String]，我想将其流式传输到外部进程并为输出返回一个 Iterable[String]。

我觉得这在编译时应该可以工作

import scala.sys.process._

object PipeUtils {
  implicit class IteratorStream(s: TraversableOnce[String]) {
    def pipe(cmd: String) = s.toStream.#>(cmd).lines
    def run(cmd: String) = s.toStream.#>(cmd).!
  }
}

但是，Scala 尝试执行 s 的内容而不是将它们传递给标准 in。谁能告诉我我做错了什么？

更新:

我认为我最初的问题是 s.toStream 被隐式转换为 ProcessBuilder 然后执行。这是不正确的，因为它是流程的输入。

我想出了以下解决方案。这感觉很老套而且是错误的，但它现在似乎有效。我不是把它写成一个答案，因为我觉得答案应该是一行而不是这个巨大的东西。

object PipeUtils {

  /**
   * This class feels wrong.  I think that for the pipe command it actually loads all of the output
   * into memory.  This could blow up the machine if used wrong, however, I cannot figure out how to get it to
   * work properly.  Hopefully http://stackoverflow.com/questions/28095469/stream-input-to-external-process-in-scala
   * will get some good responses.
   * @param s
   */
  implicit class IteratorStream(s: TraversableOnce[String]) {

    val in = (in: OutputStream) => {
      s.foreach(x => in.write((x + "\n").getBytes))
      in.close
    }

    def pipe(cmd: String) = {
      val output = ListBuffer[String]()
      val io = new ProcessIO(in,
      out => {Source.fromInputStream(out).getLines.foreach(output += _)},
      err => {Source.fromInputStream(err).getLines.foreach(println)})

      cmd.run(io).exitValue
      output.toIterable
    }

    def run(cmd: String) = {
      cmd.run(BasicIO.standard(in)).exitValue
    }
  }
}

编辑

这样做的动机来自于在 RDD 上使用 Spark 的 .pipe 函数。我想在我的本地代码中使用完全相同的功能。

Answer 1

假设 scala 2.11+，您应该按照@edi 的建议使用 lineStream。原因是当它可用时您会得到一个流式响应，而不是批处理响应。假设我有一个 shell 脚本 echo-sleep.sh:

#/usr/bin/env bash
# echo-sleep.sh
while read line; do echo $line; sleep 1; done

我们想使用如下代码从 scala 中调用它:

import scala.sys.process._
import scala.language.postfixOps
import java.io.ByteArrayInputStream

implicit class X(in: TraversableOnce[String]) {
  // Don't do the BAOS construction in real code.  Just for illustration.
  def pipe(cmd: String) = 
    cmd #< new ByteArrayInputStream(in.mkString("\n").getBytes) lineStream
}

然后如果我们像这样进行最后的调用:

1 to 10 map (_.toString) pipe "echo-sleep.sh" foreach println

序列中的数字每 1 秒出现在 STDOUT 上。如果像示例中那样缓冲并转换为 Iterable，您将失去这种响应能力。