Spark pipe PHP 实例

package test

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object PipeTest

def main(args: Array[String]) {
val sparkConf = new SparkConf().setAppName("pipe Test")
val sc = new SparkContext(sparkConf)
val a = sc.parallelize(1 to 9 , 3)
val result = a.pipe("php /home/gt/spark/bin/test.php").collect()//所以这里决定每台机子都装好php
result.foreach { x => println("!!!!!"+x) }
sc.stop()
}

}<?php
$in = fopen('php://stdin','r');
for ($x=0; $x<3; $x++){
$line += fgets($in);
}
echo $line;
echo " Hello world!";
?>

输出：
!!!!!6 Hello world!
!!!!!15 Hello world!
!!!!!24 Hello world!