根据第二个 rdd 的值过滤一个 rdd

Question

我有两个 rdd，我想用另一个的值过滤一个。

每个rdd的几个实例如下：

rdd1 = [((address1, date1),1), ((address5, date2),1), ((address1, date2),1), ((address2,date3),1)]
rdd2 = [(address1,1), (address1,1), (address2, 1), (address1, 1)]

所需的输出将是：

joined_rdd = [((address1, date1),1),((address1, date2),1),((address2,date3),1)]

所以基本上我想将元组保留在 rdd1 中，如果该元组中的地址值存在于 rdd2 中。

Answer 1

进行连接并丢弃 rdd2 中的所有内容：

rdd1 = sc.parallelize([(('address1', 'date1'),1), (('address5', 'date2'),1), (('address1', 'date2'),1), (('address2','date3'),1)])
rdd2 = sc.parallelize([('address1',1), ('address1',1), ('address2', 1), ('address1', 1)])

result_rdd = (rdd1.keyBy(lambda x: x[0][0])
                  .join(rdd2.map(lambda x: x[0])
                            .keyBy(lambda x: x)
                            .distinct())
                  .map(lambda x: x[1][0]))

result_rdd.collect()
[(('address2', 'date3'), 1), (('address1', 'date1'), 1), (('address1', 'date2'), 1)]

根据第二个 rdd 的值过滤一个 rdd

Filter an rdd depending on values of a second rdd

python

apache-spark

rdd

pyspark