Compare data in two RDD in spark

问题:

I am able to print data in two RDD with the below code.

usersRDD.foreach(println)
empRDD.foreach(println)

I need to compare data in two RDDs. How can I iterate and compare field data in one RDD with field data in another RDD. Eg: iterate the records and check if name and age in userRDD has a matching record in empRDD, if no put in separate RDD.

I tried with userRDD.substract(empRDD) but it was comparing all the fields.

回答1:

You need to key the data in each RDD so that there is something to join records on. Have a look at groupBy for example. Then you join the resulting RDDs. For each key, you get the matching values in both. If you are interested in finding the unmatched keys, use leftOuterJoin, like this:

// Returns the entries in userRDD that have no corresponding key in empRDD.
def nonEmp(userRDD: RDD[(String, String)], empRDD: RDD[(String, String)]) = {
  userRDD.leftOuterJoin(empRDD).collect {
    case (name, (age, None)) => name -> age
  }
}

回答2:

Of course the above solutions are complete and correct! Just one proposal , if and only if the RDDs are synchronized(Same rows have the same keys). You can use a distributed solution and exploit parallelism by using only spark transformations via the following tested solution:

def distrCompare(left: RDD[(Int,Int)], right: RDD[(Int,Int)]): Boolean = {
  val rdd1 = left.join(right).map{case(k, (lv,rv)) => (k,lv-rv)}
  val rdd2 = rdd1.filter{case(k,v)=>(v!=0)}
  var equal = true;
  rdd2.map{
    case(k,v)=> if(v!=0) equal = false
  }
  return equal
}

You can choose the number of partitions in "join".