将项目发送到特定分区

Question

我正在寻找一种将结构发送到预先确定的分区的方法，以便其他 RDD 可以使用它们

假设我有两个键值对的 RDD

val a:RDD[(Int, Foo)]
val b:RDD[(Int, Foo)]

val aStructure = a.reduceByKey(//reduce into large data structure)
b.mapPartitions{
    iter =>
         val usefulItem = aStructure(samePartitionKey)
         iter.map(//process iterator) 
}

我该如何设置分区，以便为 mapPartition 提供我需要的特定数据结构，但我不会有发送所有值的额外开销（如果我要创建一个广播变量）。

我一直认为将对象存储在 HDFS 中，但我不确定这是否是次优解决方案。

我目前正在探索的另一个想法是，是否有某种方法可以创建自定义分区或分区器来保存数据结构（尽管这可能会变得太复杂并成为问题）

感谢您的帮助！

编辑：

Pangea 提出了一个很好的观点，我应该提供更多细节。本质上，我得到了 SparseVectors 的 RDD 和倒排索引的 RDD。倒排索引对象非常大。

我希望在向量的 RDD 中做一个 MapPartitions，我可以将每个向量与倒排索引进行比较。问题是我每个分区只需要一个倒排索引对象，并且进行连接会导致我拥有该索引的大量副本。

val vectors:RDD[(Int, SparseVector)]

val invertedIndexes:RDD[(Int, InvIndex)] = a.reduceByKey(generateInvertedIndex)
vectors:RDD.mapPartitions{
    iter =>
         val invIndex = invertedIndexes(samePartitionKey)
         iter.map(invIndex.calculateSimilarity(_))
         ) 
}

Answer 1

Partitioner 是一个函数，给定一个通用元素，它将返回它属于哪个分区。它还决定分区的数量。 reduceByKey 有一种形式，它将分区器作为参数。 如果我正确理解您的问题，您希望在进行缩减时对数据进行分区。 看例子：

// create example data
val a =sc.parallelize(List( (1,1),(1,2), (2,3),(2,4) ) )
// create simple sample partitioner - 2 partitions, one for odd
// one for even key.hashCode. You should put your partitioning logic here
val p = new Partitioner { def numPartitions: Int = 2; def getPartition(key:Any) = key.hashCode % 2 }
// your reduceByKey function. Sample: just add
val f = (a:Int,b:Int) => a + b
val rdd = a.reduceByKey(p, f)
// here your rdd will be partitioned the way you want with the number
// of partitions you want
rdd.partitions.size
res8: Int = 2

rdd.map() .. // go on with your processing