3

我有两个 spark RDD，A 有 301,500,000 行，B 有 1,500,000 行。B 中的那 150 万行也都出现在 A 中。我想要这两个 RDD 之间的设置差异，这样我返回 A 的 300,000,000 行，而 B 中的 1,500,000 行不再存在于 A 中。

我不能使用 Spark DataFrames。

这是我现在使用的系统。这些 RDD 有主键。我在下面所做的是创建一个（收集的）出现在 B 中的主键列表，然后遍历 A 的主键以查找那些未出现在 B 主键列表中的主键。

a = sc.parallelize([[0,"foo",'a'],[1,'bar','b'],[2,'mix','c'],[3,'hem', 'd'],[4,'line','e']])
b = sc.parallelize([[1,'bar','b'],[2,'mix','c']])
b_primary_keys = b.map(lambda x: x[0]).collect()  # since first col = primary key


def sep_a_and_b(row):
    primary_key = row[0]
    if(primary_key not in b_primary_keys):
        return(row)


a_minus_b = a.map(lambda x: sep_a_and_b(x)).filter(lambda x: x != None)

现在，这适用于这个示例问题，因为 A 和 B 很小。但是，当我使用我的真实数据集 A 和 B 时，这是不成功的。有没有更好（更并行）的方法来实现这个？

4

1 回答 1

6

这似乎是您可以解决的问题subtractByKey

val filteredA = a.subtractByKey(b)

要更改为键值：

val keyValRDD = rdd.map(lambda x: (x[:1],x[1:]))

*请注意，我的 python 很弱，可能有更好的方法来拆分值

于 2015-09-29T16:05:24.547 回答