PySpark中的聚合

Question

我的数据格式是

UserId\tItemId:Score,ItemId:Score
UserId\tItemId:Score,ItemId:Score,ItemId:Score

等等……

我想通过减去平均值并除以标准差来规范化分数。 我的数据存储在S3上，大约有300个文件，每个文件大约30Mb。 我正在使用PySpark。 这是我尝试的代码：

lines = sc.textFile("s3n://data-files/clustering")

Itr1 = lines.map(lambda x:str(x))

Itr1.take(3)

['1\t1:0.1,2:0.2', '2\t3:0.4,4:0.6', '3\t5:0.8,6:0.1']


Itr2 = Itr1.map(lambda x: x.split("\t"))

Itr2.take(3)

[['1', '1:0.1,2:0.2'], ['2', '3:0.4,4:0.6'], ['3', '5:0.8,6:0.1']]

ItemRecScore = Itr2.map(lambda x:[x[1]])

ItemRecScore.take(3)

[['1:0.1,2:0.2'], ['3:0.4,4:0.6'], ['5:0.8,6:0.1']]

ItemRecScoreClean = ItemRecScore.map(lambda x: x[0].replace(':',' '))

ItemRecScore.take(3)

['1 0.1,2 0.2', '3 0.4,4 0.6', '5 0.8,6 0.1']

1) 我该如何提取出分数，以便我可以调用mean()和stdev()来计算这些参数。

2) 我该如何转换分数？

我对PySpark还很陌生，如果这看起来是个显而易见的简单任务，我很抱歉。 如果有任何指导或教程能教我如何在PySpark中处理和聚合数据，那就太好了。

Answer 1

可能是这个 ...

import numpy as np
itr1 = sc.textFile('hdfs:///user/rkarra777/data/mean.txt')
itr2 = itr1.map(lambda x : x.split('\\t')[1]).map(lambda x : x.split(','))
itr3 = itr2.map(lambda x : [float(str((k.split(':')[1]))) for k in x ])
itr4 = itr3.map(lambda x : np.mean([item for item in x]))

Answer 2

因为你每一行输入都有多个分数，我们可以用 flatMap 来提取每一行中的所有项目ID和分数，这样就能得到一个RDD，其中每个元素都是一个项目和分数的值。接下来，我们可以只提取分数，并把它转换成浮点数，这样PySpark的数值方法就可以对它进行处理了。完成这些后，我们只需要在RDD上调用stats()，就能得到你感兴趣的信息。

inputData = sc.textFile(inputFile)
idScores = inputData.flatMap(lambda x: x.split("\t")[1].split(","))
scores = idScores.map(lambda x: float(x.split(":")[1]))
print scores.stats()