来自 DataFrame 的 RowMatrix 包含空值

Question

我有一个DataFrame 相对于电影的用户评分（从 1 到 5）。为了获得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 的细微差别对于正确的问题定义是次要的 - 在稀疏数据的情况下，相关性到底是什么意思。

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

总体而言，您可以通过两种方式解决此问题：

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

  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"))
  

  similarity 实现所需的相似度指标。

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

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

注意

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