在Spark Stream中保存PairRdd时出错[重复]

我试图在spark流中保存我的Pair Rdd但在最后一步保存时出错。

这是我的示例代码

def main(args: Array[String]) {

val inputPath = args(0)
val output = args(1)
val noOfHashPartitioner = args(2).toInt

println("IN Streaming ")
val conf = new SparkConf().setAppName("Simple Application").setMaster("local[*]")
val sc = new SparkContext(conf)
val hadoopConf = sc.hadoopConfiguration;
//hadoopConf.set("fs.s3.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")

val ssc = new org.apache.spark.streaming.StreamingContext(sc, Seconds(60))
val input = ssc.textFileStream(inputPath)

val pairedRDD = input.map(row => {
  val split = row.split("\\|")
  val fileName = split(0)
  val fileContent = split(1)
  (fileName, fileContent)
})
import org.apache.hadoop.io.NullWritable
import org.apache.spark.HashPartitioner
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat

class RddMultiTextOutputFormat extends MultipleTextOutputFormat[Any, Any] {
  override def generateActualKey(key: Any, value: Any): Any = NullWritable.get()
  override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = key.asInstanceOf[String]
}
//print(pairedRDD)

pairedRDD.partitionBy(new HashPartitioner(noOfHashPartitioner)).saveAsHadoopFile(output, classOf[String], classOf[String], classOf[RddMultiTextOutputFormat], classOf[GzipCodec])

ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate

}
我在保存时已经到了最后一步。一定在这里遗漏了什么，得到的错误就像

value partitionBy不是org.apache.spark.streaming.dstream.DStream [（String，String）]的数据