title: EPIPE (Broken pipe) with spark pipe action
body: <p>I am trying to run a spark application with yarn. The application uses the <code>pipe()</code> action to run a local php program. The wired thing is that every time the PHP process is started, it will receive a <code>SIGPIPE</code> signal about 1 minute later (after a few records processed successfully) and get terminated.</p>
  
  <p>PS:  the same program could run smoothly on the standalone  os env, but goes wrong in production env ,which is a cluster.</p>
  
  <p>Can anyone help what's the possible cause ? thanks.</p>
  
  <pre class="lang-none prettyprint-override"><code>write(1, "success\te10489280713f60626d15d20"..., 34564) = 26372
  --- SIGPIPE (Broken pipe) @ 0 (0) ---
  write(1, "ang__meta\":{\"value\":\"empty\",\"gro"..., 8192) = -1 EPIPE &gt;(Broken pipe)
  --- SIGPIPE (Broken pipe) @ 0 (0) ---
  --- SIGTERM (Terminated) @ 0 (0) ---
  </code></pre>
  
  <p>code is as below:<br>
  Scala side:</p>
  
  <pre><code>def main(args: Array[String]): Unit = {
      if (args.length &lt; 2) {
        sys.exit(1)
      }
      val dt_src = args(0)
      val limit = args(1)
      var sql = s"SELECT * FROM db_dw.table_input_json  WHERE dt='${dt_src}'"
      val limit_str = s" LIMIT ${limit}"
      if (limit != "0") {
          sql = sql + limit_str
      }
  val df = sqlContext.sql(sql)
  var rdd = df.map(r =&gt;  r.getAs[String]("json_str")).pipe("/home/work/software/php56/bin/php ./yiic php_file index"
  }
  </code></pre>
  
  <p>PHP side: </p>
  
  <pre><code>public function actionIndex()
  {
  while ($line = fgets(STDIN)) {
           list($success, $json_str,$message) = $this-&gt;_handle($line);
           if(!$success) {
               echo "fail" . "\t"  . $message . "\n";
               continue;
           }
            echo "success" . "\t"  . $json_str . "\n"
  }
  </code></pre>