为grep -i shell命令抛出rdd.pipe抛出java.lang.IllegalStateException吗？

Question

我正在运行在RDD Spark操作中使用管道的代码：

我尝试过的以下代码段：

//PIPE - run a external shell script in spark

val x = sc.parallelize(Array("A", "Ba", "C", "AD"))
val y = x.pipe("grep -i A")
println(x.collect())
println(y.collect())

但是我得到了：

org.apache.spark.SparkException：由于阶段失败而导致作业中止：61.0阶段中的任务0失败1次，最近一次失败：61.0阶段中的任务0.0丢失（TID 592，本地主机，执行程序驱动程序）：java.lang。 IllegalStateException：子进程以状态1退出。命令运行：grep -i用于运行上述代码段。

是否可以通过grep -i运行pipe命令？

我尝试调用.sh脚本，并且该脚本正在运行，但是我想将其作为shell命令运行。Reference

Answer 1

这是因为数据已分区。即使您在.sh文件中使用与您提到的命令相同的命令，也会出现相同的错误。如果将RDD重新分区到一个分区，它将正常工作：

val y = x.repartition(1).pipe("grep -i A")

根据official documentation：

pipe(command, [envVars])：

通过shell命令将RDD的每个分区Perl或bash脚本。将RDD元素写入流程的标准输入和行输出到其stdout的结果将作为字符串的RDD返回。

当您使用grep命令时，您不能将每一行彼此独立地处理，因为如果一个元素失败，则该行存在。