title: RDD pipe one external process per line
body: <p>I am currently using apache spark as a workflow system, in standalone mode. My context is that I have a csv file where each line describes the parameters of one instance of my simulation.</p>
  
  <p>My input data looks like that:</p>
  
  <pre class="lang-csv prettyprint-override"><code>id,socket,type,platform,workload
  00001,28001,fcfs,platform.xml,workload1.json
  00002,28002,fcfs,platform.xml,workload2.json
  </code></pre>
  
  <p>So I would call my simulator with
  </p>
  
  <pre><code>simulator --tcp-port 28001 -arg1 fcfs -arg2 platform.xml -arg3 workload.json
  </code></pre>
  
  <pre class="lang-scala prettyprint-override"><code>// Simple user-defined function to call
  // sc.addFile to all file of the local folder
  addFolderToContext(sc, input_folder) 
  // Loading the csv from my file system
  val inputsRDD = spark.read.format("csv")
     .schema(schema) 
     .option("header", "true")
     .load(input_folder + "/inputs.csv")
  
  // Some preprocessing filtering
  val simulationsFCFS = inputsRDD.filter($"type" === "fcfs")
     .collect()
     .map(r =&gt; new String(r.mkString("",",","")))
     .take(2) // Only two for testing
  
  sc.parallelize(simulationsFCFS)
      .map(r =&gt; r)
      // Since each line is a simulation, 
      // I want to have one forked process per line 
      // .repartition(simulationsFCFS.size) // I try this, but it does not work.
      .pipe(Seq(scriptFcfs),
       Map(),
       null,
       null,
       false,
       1024, 
       Codec.defaultCharsetCodec.name) 
      .collect()
  </code></pre>
  
  <p>I have been able to use the RDD.pipe function to call my external
  program. But for the moment, I am forced to have a wrapper script that loops
  through /dev/stdin and call my wrapper for each line. And this is because the pipe function applies all the line from one partition to the same subprocess of my program, and not per "line".</p>
  
  <p>Whereas, I can achieve good parallelism, which was one of my requirment (to be able to parallelize my simulations), I don't have much control over a specific instance. What I would like to have is one call to my external process per line of my entry csv. The benefits, would be to be able to keep track the failed simulations, because they would have been a standard spark task. Can someone helps me to achieve it please?</p>
  
  <p>Thanks very much.</p>