您只为输入指定了2个分区，为输出指定了1个分区。这样做的一个影响是严重限制了这些操作的并行性。为什么需要这些？
与其计算50000个过滤过的RDD（速度也很慢），不如直接按键分组？我知道你想把它们输出到不同的目录，但这确实造成了这里的瓶颈。有没有另一种方法可以简单地让您读取（键、值）结果？

我认为这种方法应该类似于通过键spark-one-spark作业写入多个输出。分区号与目录号无关。要实现它，您可能需要用自定义版本覆盖generatefilenameforkeyvalue以保存到不同的目录。
关于可伸缩性，这不是spark的问题，而是hdfs的问题。但不管你怎么实现，只要要求不改变，都是不可避免的。但我认为hdfs可以处理50000个文件处理程序

def writeLines(iterator: Iterator[(String, String)]) = {
  val writers = new mutalbe.HashMap[String, BufferedWriter] // (key, writer) map
  try {
  while (iterator.hasNext) {
    val item = iterator.next()
    val key = item._1
    val line = item._2
    val writer = writers.get(key) match {
      case Some(writer) => writer
      case None =>
        val path = arg(1) + key
        val outputStream = FileSystem.get(new Configuration()).create(new Path(path))
        writer = new BufferedWriter(outputStream)
    }
    writer.writeLine(line)
    } finally {
    writers.values.foreach(._close())
    }
}

val inputData = sc.textFile()    
val keyValue = inputData.map(line => (key, line))
val partitions = keyValue.partitionBy(new MyPartition(10))    
partitions.foreachPartition(writeLines)

class MyPartitioner(partitions: Int) extends Partitioner {
    override def numPartitions: Int = partitions

    override def getPartition(key: Any): Int = {
        // make sure lines with the same key in the same partition 
        (key.toString.hashCode & Integer.MAX_VALUE) % numPartitions 
    }
}