Spark Pipe调用外部程序

spark 中，有种特殊的Rdd，即pipedRdd，提供了调用外部程序如基于CUDA的C++程序，使其能够更快的进行计算。caffe on spark 和tensorflow on spark 也是基于此机制，那么，spark 和 外部程序是怎么交互的呢？ 下面通过一个简单的例子验证。

步骤1：创建外部脚本

#!/bin/sh
echo "Running shell script"
while read LINE; do
echo ${LINE}!
done

步骤2：spark rdd 调用

val data = sc.parallelize(List("hi","hello","how","are","you"))
val scriptPath = "/root/echo.sh"
val pipeRDD = dataRDD.pipe(scriptPath)
pipeRDD.collect()

查看运行结果，发现为：

Array[String] = Array(Running shell script, hi!, Running shell script, hello!, Running shell script, how!, Running shell script, are!, you!)

其中，Running shell script只出现了4次，rdd的count为9，可见有两次共享了一个外部进程。在此推断和RDD并行度有关。

步骤3：调整并行度

dataRDD.repartition(5)

结果为：

Array[String] = Array(Running shell script, are!, Running shell script, hi!, hello!, you!, Running shell script, Running shell script, Running shell script, how!)

Running shell script只出现了5次，rdd的count为10了。

结论

rdd pipe 每个分区，启动一次外部程序，输入通过stdin传入，结果通过stdout传出。