сравнивая данные в двух RDDS

Question

rdd1:

(m1,p1)
(m1,p2)
(m1,p3)
(m2,p1)
(m2,p2)
(m2,p3)
(m2,p4)

rdd2:

(m1,p1)
(m1,p2)
(m1,p3)
(m2,p1)
(m2,p2)
(m2,p3)

У меня есть два числа rdd1 и rdd2 Я хочу сравнить оба числа и распечататьразличия, т.е. (m2,p4), которого нет в rdd2.

Я пытался rdd1.substractbykey(rdd2) и rdd1.substract(rdd2) Я не получаю никаких данных, пожалуйста, помогите

Answer 1

Вы можете использовать full outer join в данных:

def find_not_null(row):
     if(row['col1'] is None):
         return (row['col3'], row['col4'])
     else:
         return (row['col1'], row['col2'])

diff_rdd = rdd1.toDF(['col1', 'col2']). \
   join(rdd1.toDF(['col3', 'col4']), \ 
        col('col1') == col('col2') and col('col3') == col('col4'), \
        'full_outer'). \
   filter(lambda x: x['col1'] is None or x['col3'] is None).rdd. \
   map(find_not_null)

Answer 2

Если вам действительно нужно RDD с, то вы могли бы получить свой результат, используя subtract и union.

Если предположить, что вас интересуют различия с обеих сторон, это сработает:

val left = sc.makeRDD(Seq(("m1","p1"), ("m1","p2"), ("m1","p3"), ("m2","p1"), ("m2","p2"), ("m2","p3"), ("m2","p4")))
val right = sc.makeRDD(Seq(("m1","p1"), ("m1","p2"), ("m1","p3"), ("m2","p1"), ("m2","p2"), ("m2","p3"), ("m3","p1")))

val output = left.subtract(right).union(right.subtract(left))
output.collect() // Array[(String, String)] = Array((m2,p4), (m3,p1))

С другой стороны, если вы не против сохранить «полное внешнее соединение» в памяти, вы можете добиться того же, используя cogroup:

val output = left.cogroup(right).flatMap { case (k, (i1, i2)) => 
  val s1 = i1.toSet
  val s2 = i2.toSet
  val diff = (s1 diff s2) ++ (s2 diff s1)
  diff.toList.map(k -> _)
}
output.collect() // Array[(String, String)] = Array((m2,p4), (m3,p1))

Answer 3

Попробуйте это -

rdd1:

(m1,p1) (m1,p2) (m1,p3) (m2,p1) (m2,p2) (m2,p3) (m2,p4)

rdd2:

(m1,p1) (m1,p2) (m1,p3) (m2,p1) (m2,p2) (m2,p3)