0
私はいつも私のRDDの行動を試してこのエラーが発生し続ける&それを修正する方法は？flatMapの適切な使用
/databricks/spark/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
317 raise Py4JJavaError(
318 "An error occurred while calling {0}{1}{2}.\n".
--> 319 format(target_id, ".", name), value)
320 else:
321 raise Py4JError(
私は上の問題がflatMapであることを示し、そのratingByUser、アクションを行うことができ、最後RDDである把握することを試みました。
私はCSVを（userID、movieID、rating）で受け取り、ユーザーIDごとにmovieIDと評価を一意に組み合わせて作成したいと考えていますが、異なるユーザーが同じmovieIDのペアを生成できます。このCSVのための例：
1,2000,5
1,2001,2
1,2002,3
2,2000,4
2,2001、
キー（2000,2001）、値（5,2,1）
キー（2000,2002）、値（：
2,2004,5
1私はRDDしたいです5,3,1）
キー（2001,2002）、値（2,3,1）
キー（2000,2001）、値（4,1,1）
キー（2000,2004）、値（4,5,1）
キー（2001,2004）、値（1,5,1）
# First Map function - gets line and returns key(userID) value(movieID,rating)
def parseLine(line):
fields=line.split(",")
userID=int(fields[0])
movieID=int(fields[1])
rating=int(fields[2])
return userID, (movieID,rating)
# Function to create movie unique pairs with ratings
# all pair start with the lowest ID
# returns key (movieIDj,movieIDi) & value (rating-j,rating-i,1)
# the 1 in value is added in order to count number of ratings in the reduce
def createPairs(userRatings):
pairs=[]
for i1 in range(len(userRatings[1])-1):
for i2 in range(i1+1,len(userRatings[1])):
if userRatings[i1][0]<userRatings[1][i2][0]:
pairs.append(((userRatings[1][i1][0],userRatings[1][i2][0]),(userRatings[1][i1][1],userRatings[1][i2][1],1)))
else:
pairs.append(((userRatings[1][i2][0],userRatings[1][i1][0]),(userRatings[1][i2][1],userRatings[1][i1][1],1)))
return pairs
# Create SC object from the ratings file
lines = sc.textFile("/FileStore/tables/dvmlbdnj1487603982330/ratings.csv")
# Map lines to Key(userID),Value(movieID,rating)
movieRatings = lines.map(parseLine)
# Join all rating by same user into one key
# (UserID1,(movie1,rating1)),(UserID1,(movie2,rating2)) --> UserID1,[(movie1,rating1),(movie2,rating2)]
ratingsPerUser = movieRatings.groupByKey()
# activate createPairs func
# We use flatMap, since each user have different number of ratings --> different number pairs
pairsOfMovies = ratingsPerUser.flatMap(createPairs)