因为您的密钥不能保证在一个黑名单中，所以我要做的是生成一个flatmap，在这里您将（key，value）rdd更改为（blacklist key，value），其中数据可能重复，然后使用multipletextoutputformat根据“blacklist/key”写入文件路径。类似于此：

// Simple MultipleTextOutputFormat to output filename by key
// Directory separators are handled gracefully
class BlacklistMultipleTextOutputFormat 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]
}

myKvRdd = ... // your code to generate your RDD of (key, value)
blacklist:Map[String, Set[String]] = ... // your code to generate (key -> Set[Blacklist]) 

// Change it to (blacklist/key, value) where there is one row per key per blacklist that it blongs to
blacklisted = myKvRdd.flatMap(kv => blacklist.get(kv._1).map(b => (b + '/' + kv._1, kv._2)

// 100 partitions is pretty arbitrary, but it should be sufficiently large so that you don't OOM your executors
blacklisted.partitionBy(new HashPartitioner(100))
    .saveAsHadoopFile("absoluteOutputPath", classOf[String], classOf[String],
        classOf[BlacklistMultipleTextOutputFormat])

然后您可以在rdd的一个过程中生成输出。假设这个rdd比原来的要大，这个工作流似乎更适合spark的优化。