如何根據密鑰從PairRDD獲取新的RDD

Question

在我的Spark應用程序中，我使用了一個數據量很大的JavaPairRDD<Integer, List<Tuple3<String, String, String>>>。

而我的要求是，我需要一些基於密鑰的Large PairRDD的其他RDDs JavaRDD<Tuple3<String, String, String>>。

Answer 1

我不知道Java的API，但這裏是你會怎麼做，在斯卡拉（在spark-shell）：

def rddByKey[K: ClassTag, V: ClassTag](rdd: RDD[(K, Seq[V])]) = { 
    rdd.keys.distinct.collect.map { 
    key => key -> rdd.filter(_._1 == key).values.flatMap(identity) 
    } 
} 

你必須filter每個鍵與flatMap扁平化List秒。我不得不提的是，這不是一個有用的操作。如果您能夠構建原始RDD，那意味着每個List都足夠小以適應內存。所以我不明白你爲什麼想把它們變成RDD。