タプルのRDDでpysparkデータフレーム列を更新する

Question

私はユーザー情報が文字列であるデータを扱っています。これらの文字列に固有の整数値を割り当てたいと思います。

私はこのスタックオーバーフローポストhereにいくらか従っていました。私は最終的にALSモデルを実行されて何をしたいのか

data = data.map(lambda x: Rating(int(user.lookup(x[0])), int(x[1]), float(x[2]))) 

をした、その後

user = data.map(lambda x:x[0]).distinct().zipWithUniqueId() 

、これまでのところ、私：私はタプルのRDDを持ってするには、以下の式を使用していますこのエラーメッセージが表示されています。

例外：アクションまたはトランスフォーメーションからRDDをブロードキャストするか、RDDを参照しようとしているようです。

データ型が何とか間違っていると思いますが、これを修正する方法がわかりません。

Answer 1

lookupリンクされた回答で示唆されたアプローチは単に無効です。 Sparkはネストされたアクションや変換をサポートしていないので、をmapの中に呼び出すことはできません。データが大きいのであればあなたは、単にjoinと再構築することができます検索に標準のPython dictを使用して処理する：

は

from operator import itemgetter 
from pyspark.mllib.recommendation import Rating 

data = sc.parallelize([("foo", 1, 2.0), ("bar", 2, 3.0)]) 

user = itemgetter(0) 

def to_rating(record): 
    """ 
    >>> to_rating((("foobar", 99, 5.0), 1000)) 
    Rating(user=1000, product=99, rating=5.0) 
    """ 
    (_, item, rating), user = record 
    return Rating(user, item, rating) 

user_lookup = data.map(user).distinct().zipWithIndex() 

ratings = (data 
    .keyBy(user) # Add user string as a key 
    .join(user_lookup) # Join with lookup 
    .values() # Drop keys 
    .map(to_rating)) # Create Ratings 

ratings.first() 
## Rating(user=1, product=1, rating=2.0)