在Pyspark中连接两个具有多个pule值的rdd并添加额外的值？

Question

我创建了如下2RDD's：

rdd1 = sc.parallelize([(u'176', u'244', -0.03925566875021147), (u'28', u'244', 0.9175106515709205), (u'165', u'244', -0.3837580218245722), (u'181', u'244', 0.29145693160561503), (u'161', u'244', -0.503468718448459), (u'28', u'275', 1.1636548589189926), (u'165', u'275', -1.026158464467282), (u'181', u'275', 0.6685791983070568)])

rdd2 = sc.parallelize([(u'176', u'244'), (u'28', u'244'), (u'165', u'244'), (u'165', u'275'), (u'181', u'275'), (u'141', u'388'), (u'154', u'238')])

我的预期输出如下：

[(u'176', u'244', -0.03925566875021147,1), (u'28', u'244', 0.9175106515709205,1), (u'165', u'244', -0.3837580218245722,1), (u'181', u'244', 0.29145693160561503,0), (u'161', u'244', -0.503468718448459,0), (u'28', u'275', 1.1636548589189926,0), (u'165', u'275', -1.026158464467282,1), (u'181', u'275', 0.6685791983070568,1)]

我想加入两个RDD添加加入状态，如1或0。你知道吗

在rdd1中，第一个元组是(u'176', u'244', -0.03925566875021147)，rdd2包含 (u'176', u'244')，rdd1、rdd2的前两个元素相同，然后我的预期输出是(u'176', u'244', -0.03925566875021147,1)。你知道吗

在Rdd1:(u'181', u'275', 0.6685791983070568)和Rdd2:(u'181', u'275')的情况下也是一样的，输出将是(u'181', u'275', 0.6685791983070568,1)。你知道吗

其他情况： rdd1包含(u'181', u'244', 0.29145693160561503)，但rdd2不包含任何类似(u'181', u'244')的元组，因此预期的输出将是（u'181', u'244', 0.29145693160561503,0)

我通过创建数据帧实现了这一点，但我不想使用数据帧连接。请帮助我们如何使用rdds来实现。你知道吗

Answer 1

I want to join two rdds add join status like 1 or 0

要加入rdd，您需要pairedRdd

pairedRdd1 = rdd1.map(lambda x: ((x[0], x[1]), x[2:]))
pairedRdd2 = rdd2.map(lambda x: ((x[0], x[1]), 1))

在这里，我在pairedRdd2中填充了1，因为您的输出要求是有1来匹配rdd1中的rdd2。你知道吗

最后，对预期的输出使用leftOuterJoin和一些操作

finalRdd = pairedRdd1.leftOuterJoin(pairedRdd2).map(lambda x: tuple(list(x[0]) + list(x[1][0]) + [0 if(x[1][1] == None) else 1]))
#[('161', '244', -0.503468718448459, 0),('165', '244', -0.3837580218245722, 1),('181', '244', 0.29145693160561503, 0),('165', '275', -1.026158464467282, 1),('181', '275', 0.6685791983070568, 1),('176', '244', -0.03925566875021147, 1),('28', '275', 1.1636548589189926, 0),('28', '244', 0.9175106515709205, 1)]

我希望答案是有帮助的

Answer 2

要在rdd方法中做到这一点，您必须将rdd与您想要的列配对加入。和然后执行这个和那个的左外连接。对于其中的每个元素（k，v），生成的RDD要么包含其他元素中w的所有对（k，（v，Some（w）），要么包含其他元素中没有键k的对（k，（v，None））

 userRDD.leftOuterJoin(empRDD).collect {
        case (String, (firstrddvalue, None)) => (k,v,0)
        case (String, (firstrddvalue,secondrddvalue))=>(k,v,1)
      }