hadoop - Spark 1.2 : Write single record into multiple files (blacklisted)

Question

我尝试将 RDD 中的每条记录写入多个文件(每个黑名单一个，并按键分组)到 HDFS，并在每个文件集上应用黑名单。

首先，我将 MultipleTextOutputFormat 与 keyBy 结合使用，按记录中的字段对输出文件进行分组，效果很好。所以我的输出文件现在由一个键命名，来自记录，记录在这个文件中分组。

但我现在的问题是，我需要在输出上应用黑名单并分别保存这些输出中的每一个。我使用一个简单的过滤器做到了这一点。现在发生的情况是，应用此文件管理器会导致作业针对 x 个不同的黑名单完成 X 次。对于大量记录，这是 Not Acceptable 。即使之前在 Dataframe 上调用缓存函数。

为了弄清楚我想要什么，我的输入是一个巨大的分布式 HDFS 文件，其结构为:

ID, Name, .

我的输出目录应该是这样的:

blacklisted1/ID1.file -> Content:(ID, Name, ...)
          /ID2.file
blacklisted2/ID1.file
          /ID2.file

目标是只读取每条记录一次并为其写入这些文件集。我目前的方法是使用 forEachPartition 并通过遍历记录在 HDFS 中手动创建一个文件，但对我来说，看起来必须有一个更简单的解决方案。

Answer 1

由于不能保证您的 key 在一个黑名单中，我要做的是生成一个 flatMap，您可以在其中将 (key, value) RDD 更改为 (blacklist-key, value)，其中数据可能会重复，然后使用 MultipleTextOutputFormat 写入基于“黑名单/ 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 比原来的 RDD 大，这个工作流似乎更适合 spark 的优化。