如何将火花rdd的副本存储到另一个rdd

Question

我有两个spark rdd：RDD1：RDD [（String，String，Int）]和RDD2：RDD [（String，String，Int）]

RDD1是原始数据，而RDD2与RDD1不同

我需要创建一个RDD3，它是RDD1-RDD2

例如：

RDD1:  [("one","one",23)],[("one","one",23)],[("two","two",28)],[("one","one",23)]
RDD2:  [("one","one",23)],[("two","two",28)]

期望

RDD3:[("one","one",23)],[("one","one",23)] 

仅重复项减少了1的重复项

RD＃是仅重复项的集合，例如，如果10个事务存在，则1是唯一的，因此我应该在RDD3中收集9个事务

Answer 1

val rdd1 = sc.parallelize(List(("A", "ANT", 1), ("A", "ANT", 1), ("B", "BALL", 2), ("C", "CAT", 3), ("C", "CAT", 3))).zipWithIndex()
rdd1.collect().foreach(r => print(r))

val rdd2 = rdd1.reduceByKey((a, b) => if (a != b) a else b)
rdd2.collect().foreach(r => print(r))


val rdd3 = rdd1.subtract(rdd2).map(p => p._1)
rdd3.collect().foreach(r => print(r))

结果：

RDD1: ((A,ANT,1),0)((A,ANT,1),1)((B,BALL,2),2)((C,CAT,3),3)((C,CAT,3),4)
RDD2: ((A,ANT,1),0)((C,CAT,3),3)((B,BALL,2),2)
RDD3: (C,CAT,3)(A,ANT,1)