Spark:对RDD中的有效质量查找

Question

在Apache Spark中我有两个RDD.第一个data : RDD[(K,V)]包含键值形式的数据.第二个pairs : RDD[(K,K)]包含一组有趣的密钥对.

如何有效地构造RDDpairsWithData : RDD[((K,K)),(V,V))],使其包含来自pairskey-tuple的所有元素及其对应的值(from data)作为value-tuple？

数据的一些属性:

• 钥匙data是独一无二的
• 所有条目pairs都是唯一的
• 对于所有对(k1,k2)中pairs的保证k1 <= k2
• "对"的大小只是数据大小的常量 |pairs| = O(|data|)
• 当前数据大小(预计会增长): |data| ~ 10^8, |pairs| ~ 10^10

目前的尝试

以下是Scala中的一些示例代码:

import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext._

// This kind of show the idea, but fails at runtime.
def massPairLookup1(keyPairs : RDD[(Int, Int)], data : RDD[(Int, String)]) = {
  keyPairs map {case (k1,k2) =>
    val v1 : String = data lookup k1 head;
    val v2 : String = data lookup k2 head;
    ((k1, k2), (v1,v2))
  }
}

// Works but is O(|data|^2)
def massPairLookup2(keyPairs : RDD[(Int, Int)], data : RDD[(Int, String)]) = {
  // Construct all possible pairs of values
  val cartesianData = data cartesian data map {case((k1,v1),(k2,v2)) => ((k1,k2),(v1,v2))}
  // Select only the values who's keys are in keyPairs
  keyPairs map {(_,0)} join cartesianData mapValues {_._2}
}

// Example function that find pairs of keys
// Runs in O(|data|) in real life, but cannot maintain the values
def relevantPairs(data : RDD[(Int, String)]) = {
  val keys = data map (_._1)
  keys cartesian keys filter {case (x,y) => x*y == 12 && x < y}
}

// Example run
val data = sc parallelize(1 to 12) map (x => (x, "Number " + x))
val pairs = relevantPairs(data)
val pairsWithData = massPairLookup2(pairs, data) 


// Print: 
// ((1,12),(Number1,Number12))
// ((2,6),(Number2,Number6))
// ((3,4),(Number3,Number4))
pairsWithData.foreach(println)

尝试1

首先我尝试使用lookup函数on data,但在执行时会抛出运行时错误.它似乎self在PairRDDFunctions特质中是空的.

另外我不确定性能lookup.文档说如果RDD通过仅搜索键映射到的分区而具有已知分区器,则此操作有效地完成.这听起来像n查找最多需要O(n*|分区|)时间,我怀疑可以优化.

尝试2

这种尝试有效,但我创建了|data|^2会破坏性能的对.我不希望Spark能够优化它.

Answer 1

您的查找1不起作用,因为您无法在工作者内部执行RDD转换(在另一个转换中).

在查找2中,我认为没有必要执行完整的笛卡尔...

你可以这样做:

val firstjoin = pairs.map({case (k1,k2) => (k1, (k1,k2))})
    .join(data)
    .map({case (_, ((k1, k2), v1)) => ((k1, k2), v1)})
val result = firstjoin.map({case ((k1,k2),v1) => (k2, ((k1,k2),v1))})
    .join(data)
    .map({case(_, (((k1,k2), v1), v2))=>((k1, k2), (v1, v2))})

或者以更密集的形式:

    val firstjoin = pairs.map(x => (x._1, x)).join(data).map(_._2)
    val result = firstjoin.map({case (x,y) => (x._2, (x,y))})
        .join(data).map({case(x, (y, z))=>(y._1, (y._2, z))})

我认为你不能更有效地做到这一点,但我可能错了......