使用元组的 RDD 更新 PySpark 数据框列答案

【问题标题】：Updating a PySpark dataframe column with an RDD of tuples使用元组的 RDD 更新 PySpark 数据框列
【发布时间】：2016-04-06 17:23:58
【问题描述】：

我正在处理用户信息为字符串的数据。我想为这些字符串分配唯一的整数值。

我有点关注这个堆栈溢出帖子here。我正在使用下面的表达式来获得元组的 RDD：

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

之后，我做到了

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

我最终想要做的是在其上运行 ALS 模型，但到目前为止我一直收到此错误消息

例外：您似乎正在尝试广播 RDD 或从操作或转换中引用 RDD。

我认为数据类型有些错误，但我不知道如何解决这个问题。

【问题讨论】：

这里有 2 个问题。第一个是想要更新 DataFrame 中的值，这是不可能的！ DataFrame 是不可变的，您必须使用更新转换从现有的创建一个新的。其次，您不能将 RDD 嵌套在另一个 RDD 转换中。如果你的 RDD 很小，你可以考虑广播变量。
@eliasah 感谢您的意见。 newData = data.map(lambda x: Rating(int(user.lookup(x[0])), int(x[1]), float(x[2]))) 工作，还是我必须做像 df = sqlContext.createDataFrame(?, [cols]) 之类的东西，我不太确定如何将东西放在 ? 的位置。至于第二部分，我在哪里将 RDD 嵌套在另一个 RDD 转换中？我的数据实际上很大。
这可能有效，您需要尝试！我无法阅读评论中的代码。至于第二部分，你的用户价值是一个 RDD。所以这里是你尝试嵌套 RDD 的地方。

标签： python apache-spark pyspark apache-spark-sql rdd

【解决方案1】：

lookup 链接答案中建议的方法根本无效。 Spark 不支持嵌套操作或转换，因此您不能在 map 内调用 RDD.lookup。如果数据太大而需要使用标准 Python dict 进行查找，您可以简单地 join 并重新整形：

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)

【讨论】：