java - スパークパイプアクション付きEPIPE（ブロークンパイプ）

Question

ヤーンでスパークアプリケーションを実行しようとしています。アプリケーションはpipe()アクションを使用して、ローカルの php プログラムを実行します。PHP プロセスが開始されるたびに、SIGPIPE約 1 分後に (いくつかのレコードが正常に処理された後) シグナルを受信し、終了します。

PS: 同じプログラムは、スタンドアロンの OS 環境ではスムーズに実行できますが、クラスターである運用環境ではうまくいきません。

誰でも考えられる原因を助けることができますか? ありがとう。

write(1, "success\te10489280713f60626d15d20"..., 34564) = 26372
--- SIGPIPE (Broken pipe) @ 0 (0) ---
write(1, "ang__meta\":{\"value\":\"empty\",\"gro"..., 8192) = -1 EPIPE >(Broken pipe)
--- SIGPIPE (Broken pipe) @ 0 (0) ---
--- SIGTERM (Terminated) @ 0 (0) ---

コードは次のとおりです:
Scala 側:

def main(args: Array[String]): Unit = {
    if (args.length < 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 =>  r.getAs[String]("json_str")).pipe("/home/work/software/php56/bin/php ./yiic php_file index"
}

PHP側:

public function actionIndex()
{
while ($line = fgets(STDIN)) {
         list($success, $json_str,$message) = $this->_handle($line);
         if(!$success) {
             echo "fail" . "\t"  . $message . "\n";
             continue;
         }
          echo "success" . "\t"  . $json_str . "\n"
}