将元组元素减去其他元组-Pyspark

Question

我是使用火花和蟒蛇的新手

我有以下元组

#Money in each account
[(key), {(4.32,32.32,23.3),(3.23,32.2,21.3)}]

我要做的是，从每个元素中减去，例如（4.32,32.32,23.3），每个帐户对应的平均值，到4.32-3.23，到32.32-32.32。。。

我的想法是做到以下几点。

subtract = myRDD.reduceByKey(lambda x,y: ((x[0][0]-y[1][0]), (x[0][1]-y[1][1]), (x[0][2]-y[1][2])))

但是我不知道如何才能正确地输入索引。

我的输入文件具有以下结构。

Index,Person,Account1,Account2,Account3
0,Serge,5.958191,0.6880646,8.135345

然后我计算每行的平均值，因此例如对于所有名为Serge的Keys，我有以下数据元组。

{('Serge'),((5.958191,0.6880646,8.135345),(3.23,32.2,21.3))}

现在我想要的是减去每个元素的平均值...

例如，在本例中，我想要一个具有以下结构的元组。

{('Serge'),((5.958191-3.23,0.6880646-32.2,8.135345-21.3, 10))}

我的代码；

data = sc.textFile("myFile.csv")

data1 = data.map(lambda x: ((x.split(",")[1]),(x.split(",")[2],x.split(",")[3],x.split(",")[4]))

mean = data.mapValues( lambda x: (x,1) ) \
     .reduceByKey (lambda x , y: ((x[0][0] + y[0][0], x[0][1] + y[0][1], x[0][2] + y[0][2]), x[1] + y[1])).mapValues (lambda a: (a[0][0]/a[1],a[0][1]/a[1],a[0][2]/a[1], a[1]))

data2 = mean.join(data1)

mapValues中的最后一个[1]是获取我的项目计数。

Answer 1

如果您有平均rdd，那么您可以将rdd还原为

(u'Serge', (-5.962059049999999, 0.6719970749999999, 8.15111525))

以及读取新文件后获得的data1 rdd

(u'Serge', (3.23, 32.2, 21.3))

而你加入他们

data2 = mean.join(data1)

所以你有

(u'Serge', ((-5.962059049999999, 0.6719970749999999, 8.15111525), (3.23, 32.2, 21.3)))

现在我想要的是减去每个元素的平均值...
例如，在这种情况下，我想要一个具有以下结构的元组。
{（'Serge'），（（5.958191-3.23,0.6880646-32.2,8.135345-21.3, 10))}

我不知道这10是从哪里来的，但要进行减法，只需使用map函数

data2.map(lambda x: ((x[0]), (x[1][0][0] - x[1][1][0], x[1][0][1] - x[1][1][1], x[1][0][2] - x[1][1][2])))

所以最终的结果是

('Serge', (-9.19205905, -31.528002925000003, -13.14888475))

我希望答案有帮助