scala - 来自 DataFrame 的 RowMatrix 包含空值

Question

我有一个关于电影的用户评分(从 1 到 5)的 DataFrame。为了获取 DataFrame，其中第一列是电影 id，其余列是每个用户对该电影的评分，我执行以下操作:

val ratingsPerMovieDF = imdbRatingsDF
  .groupBy("imdbId")
  .pivot("userId")
  .max("rating")

现在，我得到一个 DataFrame，其中大多数值都是 null，因为大多数用户只评价了几部电影。

我有兴趣计算这些电影之间的相似度(基于项目的协作过滤)。

我试图使用评级列值组装一个 RowMatrix (用于使用 mllib 进行进一步的相似性计算)。但是，我不知道如何处理 null 值。

在下面的代码中，我尝试为每一行获取一个向量:

val assembler = new VectorAssembler()
  .setInputCols(movieRatingsDF.columns.drop("imdbId"))
  .setOutputCol("ratings")

val ratingsDF = assembler.transform(movieRatingsDF).select("imdbId", "ratings")

给我一​​个错误:

Caused by: org.apache.spark.SparkException: Values to assemble cannot be null.

我可以使用 .na.fill(0) 将它们替换为 0，但这会产生不正确的相关结果，因为几乎所有向量都会变得非常相似。

任何人都可以建议在这种情况下该怎么做吗？这里的最终目标是计算行之间的相关性。我正在考虑以某种方式使用 SparseVectors (忽略 null 值，但我不知道如何操作。

我是 Spark 和 Scala 的新手，所以其中一些内容可能没有什么意义。我正在努力更好地理解事物。

Answer 1

我相信您正在以错误的方式处理这个问题。处理 Spark API 的细微差别对于正确的问题定义来说是次要的 - 在稀疏数据的情况下，相关性到底意味着什么。

在明确反馈(评级)的情况下用零填充数据是有问题的，不是因为所有向量都会变得非常相似(指标的变化将由现有的评级，并且结果始终可以使用最小-最大缩放器重新缩放)，但因为它引入了原始数据集中不存在的信息。未评级的项目与评级最低的项目之间存在显着差异。

总的来说，您可以通过两种方式解决此问题:

• 您可以仅使用两个项目都具有非缺失值的条目来计算成对相似度。如果数据集相当密集，这应该可以很好地工作。它可以使用输入数据集上的自连接来表达。使用伪代码:

  imdbRatingsDF.alias("left")
    .join(imdbRatingsDF.alias("right"), Seq("userId"))
    .where($"left.imdbId" =!= $"right.imdbId")
    .groupBy($"left.imdbId", $"right.imdbId")
    .agg(simlarity($"left.rating", $"right.rating"))
  

  其中相似度实现所需的相似度度量。

• 您可以估算缺失的评级，例如使用某种集中趋势度量。使用平均值 ( Replace missing values with mean - Spark Dataframe ) 可能是最自然的选择。

  更先进的插补技术可能会提供更可靠的结果，但在分布式系统中可能无法很好地扩展。

注意

使用SparseVectors本质上等同于na.fill(0)。